Testing with pre 8.6 build

Added testing with SSL connection (#1624 )
* Added testing with SSL connection * Update CHANGELOG.md * Revert changes * Codestyle fixes * Add version restriction * Fixed issue with connecting to SSL port to check redis version * Mark tests as relay-incompatible * Codestyle fixes
2026-08-23 06:41:55 +00:00 · 2026-01-21 10:48:00 +02:00 · 2026-01-20 17:45:10 +02:00 · 2026-01-19 10:50:39 +02:00 · 2026-01-08 13:56:55 +02:00 · 2025-12-23 13:27:27 +02:00
55 changed files with 3049 additions and 249 deletions
@@ -71,6 +71,7 @@ services:
    command: ${REDIS_EXTRA_ARGS:---enable-debug-command yes --enable-module-command yes --tls-auth-clients optional --save "" --requirepass "foobar"}
    ports:
      - "6372-6377:6372-6377"
+      - "27379-27384:27379-27384"
    volumes:
      - "./dockers/cluster:/redis/work"
    profiles:
@@ -33,6 +33,7 @@ jobs:
          - '8.0'
          - '8.2'
          - '8.4'
+          - '8.6'

    steps:

@@ -40,6 +41,7 @@ jobs:
        run: |
          # Mapping of original redis versions to client test containers
          declare -A redis_clients_version_mapping=(
+            ["8.6"]="custom-21183968220-debian-amd64"
            ["8.4"]="8.4.0"
            ["8.2"]="8.2.2-pre"
            ["8.0"]="8.0.2"
@@ -109,7 +111,7 @@ jobs:
        uses: shivammathur/setup-php@v2
        with:
          php-version: ${{ matrix.php }}
-          extensions: relay
+          extensions: relay-nightly
          coverage: ${{ (matrix.php == '8.4' && matrix.redis == '8.0') && 'xdebug' || 'none' }}

      - name: Install Composer dependencies
@@ -1,5 +1,19 @@
 ## Changelog

+## Unreleased
+
+### Fixed
+- Fixed wrong `@param` annotation in `Parameters` (#1614)
+- Make `ZRANDMEMBER` prefixable (#1621)
+- Improve connection handshake by pipelining commands (#1622)
+
+### Added
+- Added retry support (#1616)
+- Added support for VRANGE command (#1623)
+
+### Maintenance
+- Added testing with SSL connection (#1624)
+
 ## v3.3.0 (2025-11-24)
 ### Added
 - Added cluster support for `XADD`, `XDEL` and `XRANGE` (#1587)
@@ -509,6 +509,43 @@ $client = new Predis\Client('tcp://127.0.0.1', [
 For a more in-depth insight on how to create new connection backends you can refer to the actual
 implementation of the standard connection classes available in the `Predis\Connection` namespace.

+### Retry exceptions
+
+You can enable automatic retry that is disabled by default, to be able to reduce the amount of
+false-positives in case of network issues. By default, we're retrying on any connection,
+timeout or socket initialization exception, but you can update the list of retry
+exceptions. For now `EqualBackoff` and `ExponentialBackoff` strategies are available,
+but you may provide your custom one. Retry may be configured with any type of communication
+(standalone node, cluster, pipeline, transaction, replication). Here's an example of
+configuration:
+
+```php
+// Standalone client
+$client = new Predis\Client([
+    'retry' => new \Predis\Retry\Retry(
+        new \Predis\Retry\Strategy\ExponentialBackoff(1000, 10000), // Base and cap configuration in microseconds
+        3                                                           // Number of retries
+    ),
+]);
+
+// Cluster configuration
+$options = [
+    'parameters' => [
+        'retry' => new \Predis\Retry\Retry(new \Predis\Retry\Strategy\ExponentialBackoff(1000, 10000), 3),
+    ],
+];
+
+$client = new Predis\Client(['tcp://host:port', 'tcp://host:port', 'tcp://host:port'], $options);
+
+$retry = new \Predis\Retry\Retry(
+    new \Predis\Retry\Strategy\ExponentialBackoff(1000, 10000),
+    3
+);
+
+// Update a list of exceptions to catch
+$retry->updateCatchableExceptions([Exception::class]);
+```
+
 ## RESP3 ##

 ### Connection ###
@@ -29,9 +29,6 @@ parameters:
    - message: "#^Access to an undefined property Predis\\\\Connection\\\\ParametersInterface\\:\\:\\$weight\\.$#"
      count: 1
      path: src/Connection/Cluster/PredisCluster.php
-    - message: "#^Variable \\$response might not be defined\\.$#"
-      count: 2
-      path: src/Connection/Cluster/RedisCluster.php
    - message: "#^Access to an undefined property Predis\\\\Connection\\\\ParametersInterface\\:\\:\\$role\\.$#"
      count: 1
      path: src/Connection/Replication/MasterSlaveReplication.php
@@ -45,6 +42,3 @@ parameters:
    - message: "#^Variable \\$response might not be defined\\.$#"
      count: 1
      path: src/Connection/Replication/MasterSlaveReplication.php
-    - message: "#^Variable \\$response might not be defined\\.$#"
-      count: 1
-      path: src/Connection/Replication/SentinelReplication.php
@@ -55,6 +55,15 @@
        <env name="USE_RELAY" value="false" />
        <env name="REDIS_STACK_SERVER_PORT" value="6479" />

+        <!-- SSL -->
+        <env name="STANDALONE_CA_CERT_PATH" value=".github/dockers/standalone/tls/ca.crt" />
+        <env name="REDIS_SSL_PORT" value="6666" />
+        <env name="CLUSTER_CA_CERT_PATH" value=".github/dockers/cluster/tls/ca.crt" />
+        <const
+            name="SSL_REDIS_CLUSTER_ENDPOINTS"
+            value="127.0.0.1:27379?password=foobar,127.0.0.1:27380?password=foobar,127.0.0.1:27381?password=foobar"
+        />
+
        <!-- Redis Cluster -->
        <!-- Only master nodes endpoints included -->
        <const
@@ -22,6 +22,7 @@ use Predis\Command\RawCommand;
 use Predis\Command\ScriptCommand;
 use Predis\Configuration\Options;
 use Predis\Configuration\OptionsInterface;
+use Predis\Connection\AggregateConnectionInterface;
 use Predis\Connection\ConnectionInterface;
 use Predis\Connection\Parameters;
 use Predis\Connection\ParametersInterface;
@@ -41,6 +42,7 @@ use Predis\Response\ServerException;
 use Predis\Transaction\MultiExec as MultiExecTransaction;
 use ReturnTypeWillChange;
 use RuntimeException;
+use Throwable;
 use Traversable;

 /**
@@ -376,12 +378,25 @@ class Client implements ClientInterface, IteratorAggregate

    /**
     * {@inheritdoc}
+     * @throws Throwable
     */
    public function executeCommand(CommandInterface $command)
    {
-        $response = $this->connection->executeCommand($command);
        $parameters = $this->connection->getParameters();

+        if ($this->connection instanceof AggregateConnectionInterface || $this->connection instanceof RelayConnection) {
+            $response = $this->connection->executeCommand($command);
+        } else {
+            $response = $parameters->retry->callWithRetry(
+                function () use ($command) {
+                    return $this->connection->executeCommand($command);
+                },
+                function () {
+                    $this->connection->disconnect();
+                }
+            );
+        }
+
        if ($response instanceof ResponseInterface) {
            if ($response instanceof ErrorResponseInterface) {
                $response = $this->onErrorResponse($command, $response);
@@ -355,6 +355,7 @@ use Predis\Command\Redis\VADD;
 * @method $this vinfo(string $key)
 * @method $this vlinks(string $key, string $elem, bool $withScores = false)
 * @method $this vrandmember(string $key, int $count = null)
+ * @method $this vrange(string $key, string $start, string $end, int $count = null)
 * @method $this vrem(string $key, string $elem)
 * @method $this vsetattr(string $key, string $elem, string|array $attributes)
 * @method $this vsim(string $key, string|array $vectorOrElem, bool $isElem = false, bool $withScores = false, int $count = null, float $epsilon = null, int $ef = null, string $filter = null, int $filterEf = null, bool $truth = false, bool $noThread = false)
@@ -367,6 +367,7 @@ use Predis\Response\Status;
 * @method array|null        vinfo(string $key)
 * @method array|null        vlinks(string $key, string $elem, bool $withScores = false)
 * @method string|array|null vrandmember(string $key, int $count = null)
+ * @method array             vrange(string $key, string $start, string $end, int $count = null)
 * @method bool              vrem(string $key, string $elem)
 * @method array             vsim(string $key, string|array $vectorOrElem, bool $isElem = false, bool $withScores = false, int $count = null, float $epsilon = null, int $ef = null, string $filter = null, int $filterEf = null, bool $truth = false, bool $noThread = false)
 * @method bool              vsetattr(string $key, string $elem, string|array $attributes)
@@ -0,0 +1,35 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Command\Redis;
+
+use Predis\Command\PrefixableCommand as RedisCommand;
+
+class VRANGE extends RedisCommand
+{
+    /**
+     * @return string
+     */
+    public function getId()
+    {
+        return 'VRANGE';
+    }
+
+    /**
+     * @param       $prefix
+     * @return void
+     */
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForFirstArgument($prefix);
+    }
+}
@@ -12,7 +12,7 @@

 namespace Predis\Command\Redis;

-use Predis\Command\Command as RedisCommand;
+use Predis\Command\PrefixableCommand as RedisCommand;
 use Predis\Command\Traits\With\WithScores;

 /**
@@ -33,4 +33,9 @@ class ZRANDMEMBER extends RedisCommand
    {
        return 'ZRANDMEMBER';
    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForFirstArgument($prefix);
+    }
 }
@@ -19,6 +19,7 @@ use Predis\Connection\Resource\Exception\StreamInitException;
 use Predis\Protocol\Parser\ParserStrategyResolver;
 use Predis\Protocol\Parser\Strategy\ParserStrategyInterface;
 use Predis\Protocol\ProtocolException;
+use Predis\TimeoutException;

 /**
 * Base class with the common logic used by connection classes to communicate
@@ -158,6 +159,20 @@ abstract class AbstractConnection implements NodeConnectionInterface
        );
    }

+    /**
+     * Helper method to handle timeout errors.
+     *
+     * @param  int                    $code
+     * @return void
+     * @throws CommunicationException
+     */
+    protected function onTimeoutError(int $code = 0): void
+    {
+        CommunicationException::handle(
+            new TimeoutException($this, $code)
+        );
+    }
+
    /**
     * Helper method to handle protocol errors.
     *
@@ -28,10 +28,14 @@ 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;
@@ -58,7 +62,7 @@ use Traversable;
 */
 class RedisCluster extends AbstractAggregateConnection implements ClusterInterface, IteratorAggregate, Countable
 {
-    private $useClusterSlots = true;
+    public $useClusterSlots = true;

    /**
     * @var NodeConnectionInterface[]
@@ -260,35 +264,31 @@ class RedisCluster extends AbstractAggregateConnection implements ClusterInterfa
     */
    private function queryClusterNodeForSlotMap(NodeConnectionInterface $connection)
    {
-        $retries = 0;
-        $retryAfter = $this->retryInterval;
+        // Backward-compatible hardcoded retry
+        $retry = new Retry(
+            new ExponentialBackoff($this->retryInterval * 1000, -1),
+            $this->retryLimit,
+            [ConnectionException::class]
+        );
+
        $command = RawCommand::create('CLUSTER', 'SLOTS');

-        while ($retries <= $this->retryLimit) {
-            try {
-                $response = $connection->executeCommand($command);
-                break;
-            } catch (ConnectionException $exception) {
-                $connection = $exception->getConnection();
-                $connection->disconnect();
+        $doCallback = function () use (&$connection, $command) {
+            return $connection->executeCommand($command);
+        };

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

-                if ($retries === $this->retryLimit) {
-                    throw $exception;
-                }
+            $this->remove($connection);

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

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

    /**
@@ -545,51 +545,42 @@ class RedisCluster extends AbstractAggregateConnection implements ClusterInterfa
     * @param string           $method  Actual method.
     *
     * @return mixed
+     * @throws Throwable
     */
    private function retryCommandOnFailure(CommandInterface $command, $method)
    {
-        $retries = 0;
-        $retryAfter = $this->retryInterval;
-
-        while ($retries <= $this->retryLimit) {
-            try {
-                $response = $this->getConnectionByCommand($command)->$method($command);
-
-                if ($response instanceof ErrorResponse) {
-                    $message = $response->getMessage();
-
-                    if (strpos($message, 'CLUSTERDOWN') !== false) {
-                        throw new ServerException($message);
-                    }
-                }
-
-                break;
-            } catch (Throwable $exception) {
-                usleep($retryAfter * 1000);
-                $retryAfter *= 2;
-
-                if ($exception instanceof ConnectionException) {
-                    $connection = $exception->getConnection();
-
-                    if ($connection) {
-                        $connection->disconnect();
-                        $this->remove($connection);
-                    }
-                }
-
-                if ($retries === $this->retryLimit) {
-                    throw $exception;
-                }
-
-                if ($this->useClusterSlots) {
-                    $this->askSlotMap();
-                }
-
-                ++$retries;
-            }
+        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]);

-        return $response;
+        $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);
+            }
+        );
    }

    /**
@@ -740,4 +731,34 @@ class RedisCluster extends AbstractAggregateConnection implements ClusterInterfa
            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();
+            }
+        }
+    }
 }
@@ -13,6 +13,8 @@
 namespace Predis\Connection;

 use InvalidArgumentException;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\NoBackoff;

 /**
 * Container for connection parameters used to initialize connections to Redis.
@@ -36,18 +38,30 @@ class Parameters implements ParametersInterface
     */
    protected $parameters;

+    /**
+     * @var bool
+     */
+    private $disabledRetry = true;
+
    /**
     * @param array $parameters Named array of connection parameters.
     */
    public function __construct(array $parameters = [])
    {
+        if (!array_key_exists('retry', $parameters)) {
+            // Retries disabled by default
+            static::$defaults['retry'] = new Retry(new NoBackoff(), 0);
+        } else {
+            $this->disabledRetry = false;
+        }
+
        $this->parameters = $this->filter($parameters + static::$defaults);
    }

    /**
     * Filters parameters removing entries with NULL or 0-length string values.
     *
-     * @params array $parameters Array of parameters to be filtered
+     * @param array $parameters Array of parameters to be filtered
     *
     * @return array
     */
@@ -195,6 +209,16 @@ class Parameters implements ParametersInterface
        return "$this->scheme://$this->host:$this->port";
    }

+    /**
+     * Returns if retries is disabled.
+     *
+     * @return bool
+     */
+    public function isDisabledRetry(): bool
+    {
+        return $this->disabledRetry;
+    }
+
    /**
     * {@inheritdoc}
     */
@@ -12,6 +12,8 @@

 namespace Predis\Connection;

+use Predis\Retry\Retry;
+
 /**
 * Interface defining a container for connection parameters.
 *
@@ -35,9 +37,11 @@ namespace Predis\Connection;
 * @property bool   $async_connect      Performs the connect() operation asynchronously.
 * @property bool   $tcp_nodelay        Toggles the Nagle's algorithm for coalescing.
 * @property bool   $client_info        Whether to set LIB-NAME and LIB-VER when connecting.
+ * @property Retry  $retry              Retry configuration
 * @property bool   $cache              (Relay only) Whether to use in-memory caching.
 * @property string $serializer         (Relay only) Serializer used for data serialization.
 * @property string $compression        (Relay only) Algorithm used for data compression.
+ * @method   bool   isDisabledRetry()   Specify if custom retry configuration was provided.
 */
 interface ParametersInterface
 {
@@ -128,44 +128,6 @@ class RelayConnection extends AbstractConnection
        }
    }

-    /**
-     * Creates a new instance of the client.
-     *
-     * @return Relay
-     */
-    private function createClient()
-    {
-        $client = new Relay();
-
-        // throw when errors occur and return `null` for non-existent keys
-        $client->setOption(Relay::OPT_PHPREDIS_COMPATIBILITY, false);
-
-        // use reply literals
-        $client->setOption(Relay::OPT_REPLY_LITERAL, true);
-
-        // disable Relay's command/connection retry
-        $client->setOption(Relay::OPT_MAX_RETRIES, 0);
-
-        // whether to use in-memory caching
-        $client->setOption(Relay::OPT_USE_CACHE, $this->parameters->cache ?? true);
-
-        // set data serializer
-        $client->setOption(Relay::OPT_SERIALIZER, constant(sprintf(
-            '%s::SERIALIZER_%s',
-            Relay::class,
-            strtoupper($this->parameters->serializer ?? 'none')
-        )));
-
-        // set data compression algorithm
-        $client->setOption(Relay::OPT_COMPRESSION, constant(sprintf(
-            '%s::COMPRESSION_%s',
-            Relay::class,
-            strtoupper($this->parameters->compression ?? 'none')
-        )));
-
-        return $client;
-    }
-
    /**
     * Returns the underlying client.
     *
@@ -15,6 +15,8 @@ namespace Predis\Connection;
 use InvalidArgumentException;
 use Predis\Command\RawCommand;
 use Predis\NotSupportedException;
+use Predis\Retry\Strategy\EqualBackoff;
+use Predis\Retry\Strategy\ExponentialBackoff;
 use Relay\Relay;

 class RelayFactory extends Factory
@@ -64,7 +66,7 @@ class RelayFactory extends Factory
        }

        $initializer = $this->schemes[$scheme];
-        $client = $this->createClient();
+        $client = $this->createClient($parameters);

        $connection = new $initializer($parameters, $client);

@@ -90,7 +92,7 @@ class RelayFactory extends Factory
     *
     * @return Relay
     */
-    private function createClient()
+    private function createClient(ParametersInterface $parameters)
    {
        $client = new Relay();

@@ -100,26 +102,50 @@ class RelayFactory extends Factory
        // use reply literals
        $client->setOption(Relay::OPT_REPLY_LITERAL, true);

-        // disable Relay's command/connection retry
-        $client->setOption(Relay::OPT_MAX_RETRIES, 0);
-
        // whether to use in-memory caching
-        $client->setOption(Relay::OPT_USE_CACHE, $this->parameters->cache ?? true);
+        $client->setOption(Relay::OPT_USE_CACHE, $parameters->cache ?? true);

        // set data serializer
        $client->setOption(Relay::OPT_SERIALIZER, constant(sprintf(
            '%s::SERIALIZER_%s',
            Relay::class,
-            strtoupper($this->parameters->serializer ?? 'none')
+            strtoupper($parameters->serializer ?? 'none')
        )));

        // set data compression algorithm
        $client->setOption(Relay::OPT_COMPRESSION, constant(sprintf(
            '%s::COMPRESSION_%s',
            Relay::class,
-            strtoupper($this->parameters->compression ?? 'none')
+            strtoupper($parameters->compression ?? 'none')
        )));

+        if ($parameters->isDisabledRetry()) {
+            $client->setOption(Relay::OPT_MAX_RETRIES, 0);
+        } else {
+            $client->setOption(Relay::OPT_MAX_RETRIES, $parameters->retry->getRetries());
+
+            $retryStrategy = $parameters->retry->getStrategy();
+
+            if ($retryStrategy instanceof ExponentialBackoff) {
+                $algorithm = Relay::BACKOFF_ALGORITHM_FULL_JITTER;
+                $base = $retryStrategy->getBase();
+                $cap = $retryStrategy->getCap();
+            } else {
+                $algorithm = Relay::BACKOFF_ALGORITHM_DEFAULT;
+
+                if ($retryStrategy instanceof EqualBackoff) {
+                    $base = $cap = $retryStrategy->compute(0);
+                } else {
+                    $base = $retryStrategy::DEFAULT_BASE;
+                    $cap = $retryStrategy::DEFAULT_CAP;
+                }
+            }
+
+            $client->setOption(Relay::OPT_BACKOFF_ALGORITHM, $algorithm);
+            $client->setOption(Relay::OPT_BACKOFF_BASE, $base / 1000);
+            $client->setOption(Relay::OPT_BACKOFF_CAP, $cap / 1000);
+        }
+
        return $client;
    }

@@ -22,9 +22,12 @@ use Predis\Connection\ConnectionException;
 use Predis\Connection\FactoryInterface;
 use Predis\Connection\NodeConnectionInterface;
 use Predis\Connection\ParametersInterface;
+use Predis\Connection\RelayFactory;
 use Predis\Replication\MissingMasterException;
 use Predis\Replication\ReplicationStrategy;
 use Predis\Response\ErrorInterface as ResponseErrorInterface;
+use Predis\TimeoutException;
+use Throwable;

 /**
 * Aggregate connection handling replication of Redis nodes configured in a
@@ -476,9 +479,26 @@ class MasterSlaveReplication extends AbstractAggregateConnection implements Repl
     * @param string           $method  Actual method.
     *
     * @return mixed
+     * @throws Throwable
     */
    private function retryCommandOnFailure(CommandInterface $command, $method)
    {
+        $parameters = $this->getParameters();
+
+        if (!$parameters->isDisabledRetry() && !$this->connectionFactory instanceof RelayFactory) {
+            $retry = $parameters->retry;
+            $retry->updateCatchableExceptions([MissingMasterException::class]);
+
+            return $retry->callWithRetry(
+                function () use ($command, $method) {
+                    return $this->executeCommandInternal($command, $method);
+                },
+                function (Throwable $exception) {
+                    $this->onFailCallback($exception);
+                }
+            );
+        }
+
        while (true) {
            try {
                $connection = $this->getConnectionByCommand($command);
@@ -490,38 +510,35 @@ class MasterSlaveReplication extends AbstractAggregateConnection implements Repl

                break;
            } catch (ConnectionException $exception) {
-                $connection = $exception->getConnection();
-                $connection->disconnect();
-
-                if ($connection === $this->master && !$this->autoDiscovery) {
-                    // Throw immediately when master connection is failing, even
-                    // when the command represents a read-only operation, unless
-                    // automatic discovery has been enabled.
-                    throw $exception;
-                } else {
-                    // Otherwise remove the failing slave and attempt to execute
-                    // the command again on one of the remaining slaves...
-                    $this->remove($connection);
-                }
-
-                // ... that is, unless we have no more connections to use.
-                if (!$this->slaves && !$this->master) {
-                    throw $exception;
-                } elseif ($this->autoDiscovery) {
-                    $this->discover();
-                }
+                $this->onConnectionExceptionCallback($exception);
            } catch (MissingMasterException $exception) {
-                if ($this->autoDiscovery) {
-                    $this->discover();
-                } else {
-                    throw $exception;
-                }
+                $this->onMissingMasterException($exception);
            }
        }

        return $response;
    }

+    /**
+     * Executes command against valid connection.
+     *
+     * @param  CommandInterface    $command
+     * @param  string              $method
+     * @return mixed
+     * @throws ConnectionException
+     */
+    protected function executeCommandInternal(CommandInterface $command, string $method)
+    {
+        $connection = $this->getConnectionByCommand($command);
+        $response = $connection->$method($command);
+
+        if ($response instanceof ResponseErrorInterface && $response->getErrorType() === 'LOADING') {
+            throw new ConnectionException($connection, "Redis is loading the dataset in memory [$connection]");
+        }
+
+        return $response;
+    }
+
    /**
     * {@inheritdoc}
     */
@@ -571,4 +588,84 @@ class MasterSlaveReplication extends AbstractAggregateConnection implements Repl

        return null;
    }
+
+    /**
+     * Handle connection exception.
+     *
+     * @param  ConnectionException                 $exception
+     * @return void
+     * @throws ClientException|ConnectionException
+     */
+    private function onConnectionExceptionCallback(ConnectionException $exception)
+    {
+        $connection = $exception->getConnection();
+        $connection->disconnect();
+
+        if ($connection === $this->master && !$this->autoDiscovery) {
+            // Throw immediately when master connection is failing, even
+            // when the command represents a read-only operation, unless
+            // automatic discovery has been enabled.
+            throw $exception;
+        } else {
+            // Otherwise remove the failing slave and attempt to execute
+            // the command again on one of the remaining slaves...
+            $this->remove($connection);
+        }
+
+        // ... that is, unless we have no more connections to use.
+        if (!$this->slaves && !$this->master) {
+            throw $exception;
+        } elseif ($this->autoDiscovery) {
+            $this->discover();
+        }
+    }
+
+    /**
+     * Exception handling callback.
+     *
+     * @param  Throwable $exception
+     * @return void
+     * @throws Throwable
+     */
+    private function onFailCallback(Throwable $exception)
+    {
+        if ($exception instanceof ConnectionException) {
+            $this->onConnectionExceptionCallback($exception);
+
+            return;
+        }
+
+        if ($exception instanceof MissingMasterException) {
+            $this->onMissingMasterException($exception);
+
+            return;
+        }
+
+        if ($exception instanceof TimeoutException) {
+            $connection = $exception->getConnection();
+
+            if ($connection) {
+                $connection->disconnect();
+
+                return;
+            }
+        }
+
+        throw $exception;
+    }
+
+    /**
+     * @param  MissingMasterException $exception
+     * @return void
+     * @throws ClientException
+     * @throws MissingMasterException
+     */
+    private function onMissingMasterException(MissingMasterException $exception)
+    {
+        if ($this->autoDiscovery) {
+            $this->discover();
+        } else {
+            throw $exception;
+        }
+    }
 }
@@ -16,17 +16,21 @@ 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\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;

 /**
@@ -566,7 +570,10 @@ class SentinelReplication extends AbstractAggregateConnection implements Replica
    protected function assertConnectionRole(NodeConnectionInterface $connection, $role)
    {
        $role = strtolower($role);
-        $actualRole = $connection->executeCommand(RawCommand::create('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());
@@ -710,33 +717,39 @@ class SentinelReplication extends AbstractAggregateConnection implements Replica
     */
    private function retryCommandOnFailure(CommandInterface $command, $method)
    {
-        $retries = 0;
+        $parameters = $this->getParameters();

-        while ($retries <= $this->retryLimit) {
-            try {
-                $response = $this->getConnectionByCommand($command)->$method($command);
-                if ($response instanceof Error && $response->getErrorType() === 'LOADING') {
-                    throw new ConnectionException($this->current, $response->getMessage());
-                }
-                break;
-            } catch (Throwable $exception) {
-                $this->wipeServerList();
-
-                if ($exception instanceof ConnectionException) {
-                    $exception->getConnection()->disconnect();
-                }
-
-                if ($retries === $this->retryLimit) {
-                    throw $exception;
-                }
-
-                usleep($this->retryWait * 1000);
-
-                ++$retries;
-            }
+        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]);

-        return $response;
+        $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);
    }

    /**
@@ -207,7 +207,27 @@ class Stream implements StreamInterface

        $result = fwrite($this->stream, $string);

-        if ($result === false) {
+        if ($result === false || $result === 0) {
+            $metadata = $this->getMetadata();
+
+            if ($this->eof()) {
+                throw new RuntimeException('Connection closed by peer during write', 1);
+            }
+
+            if (!is_resource($this->stream)) {
+                throw new RuntimeException(
+                    'Stream resource is no longer valid',
+                    1
+                );
+            }
+
+            if (array_key_exists('timed_out', $metadata) && $metadata['timed_out']) {
+                throw new RuntimeException(
+                    'Stream has been timed out',
+                    2
+                );
+            }
+
            throw new RuntimeException('Unable to write to stream', 1);
        }

@@ -252,6 +272,26 @@ class Stream implements StreamInterface
        }

        if (false === $string) {
+            $metadata = $this->getMetadata();
+
+            if ($this->eof()) {
+                throw new RuntimeException('Connection closed by peer during read', 1);
+            }
+
+            if (!is_resource($this->stream)) {
+                throw new RuntimeException(
+                    'Stream resource is no longer valid',
+                    1
+                );
+            }
+
+            if (array_key_exists('timed_out', $metadata) && $metadata['timed_out']) {
+                throw new RuntimeException(
+                    'Stream has been timed out',
+                    2
+                );
+            }
+
            throw new RuntimeException('Unable to read from stream', 1);
        }

@@ -90,14 +90,46 @@ class StreamConnection extends AbstractConnection
    public function connect()
    {
        if (parent::connect() && $this->initCommands) {
-            foreach ($this->initCommands as $command) {
-                $response = $this->executeCommand($command);
+            $responses = $this->sendPipeline($this->initCommands);

-                $this->handleOnConnectResponse($response, $command);
+            if ($responses[0][0] instanceof ErrorResponseInterface) {
+                // Error in HELLO command, Redis < 6.0.
+                // We need to handle it separately and re-send other commands.
+                $this->handleOnConnectResponse($responses[0][0], $responses[0][1]);
+                $responses = $this->sendPipeline(array_slice($this->initCommands, 1));
+            }
+
+            foreach ($responses as $response) {
+                $this->handleOnConnectResponse($response[0], $response[1]);
            }
        }
    }

+    /**
+     * Sends commands to the server as pipeline and returns responses.
+     *
+     * @param  CommandInterface[]     $commands
+     * @return array<int, array>
+     * @throws CommunicationException
+     */
+    protected function sendPipeline(array $commands): array
+    {
+        $serialisedCommands = '';
+
+        foreach ($commands as $command) {
+            $serialisedCommands .= $command->serializeCommand();
+        }
+
+        $this->write($serialisedCommands);
+        $responses = [];
+
+        foreach ($commands as $command) {
+            $responses[] = [$this->readResponse($command), $command];
+        }
+
+        return $responses;
+    }
+
    /**
     * {@inheritdoc}
     */
@@ -340,11 +372,14 @@ class StreamConnection extends AbstractConnection
     * @param  string|null                             $message
     * @throws RuntimeException|CommunicationException
     */
-    protected function onStreamError(RuntimeException $e, ?string $message = null)
+    protected function onStreamError($e, ?string $message = null)
    {
-        // Code = 1 represents issues related to read/write operation.
+        // Code = 1 represents issues related to read/write operation, connection broken.
        if ($e->getCode() === 1) {
            $this->onConnectionError($message);
+        } elseif ($e->getCode() === 2) {
+            // Operation has been timed out, connection not necessarily broken.
+            $this->onTimeoutError();
        }

        throw $e;
@@ -14,13 +14,16 @@ namespace Predis\Pipeline;

 use Predis\ClientException;
 use Predis\ClientInterface;
-use Predis\Connection\AggregateConnectionInterface;
+use Predis\Command\Command;
+use Predis\Command\CommandInterface;
+use Predis\CommunicationException;
 use Predis\Connection\ConnectionInterface;
 use Predis\Connection\NodeConnectionInterface;
 use Predis\Response\ErrorInterface as ErrorResponseInterface;
 use Predis\Response\ResponseInterface;
 use Predis\Response\ServerException;
 use SplQueue;
+use Throwable;

 /**
 * Command pipeline wrapped into a MULTI / EXEC transaction.
@@ -63,24 +66,18 @@ class Atomic extends Pipeline
    protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
    {
        $commandFactory = $this->getClient()->getCommandFactory();
-        $connection->executeCommand($commandFactory->create('multi'));
+        $retry = $connection->getParameters()->retry;
+        $this->executeCommandWithRetry($connection, $commandFactory->create('multi'));

-        if ($connection instanceof AggregateConnectionInterface) {
-            $this->writeToMultiNode($connection, $commands);
-        } else {
-            $this->writeToSingleNode($connection, $commands);
-        }
-
-        foreach ($commands as $command) {
-            $response = $connection->readResponse($command);
-
-            if ($response instanceof ErrorResponseInterface) {
-                $connection->executeCommand($commandFactory->create('discard'));
-                throw new ServerException($response->getMessage());
+        $retry->callWithRetry(function () use ($connection, $commands) {
+            $this->queuePipeline($connection, $commands);
+        }, function (Throwable $exception) {
+            if ($exception instanceof CommunicationException) {
+                $exception->getConnection()->disconnect();
            }
-        }
+        });

-        $executed = $connection->executeCommand($commandFactory->create('exec'));
+        $executed = $this->executeCommandWithRetry($connection, $commandFactory->create('exec'));

        if (!isset($executed)) {
            throw new ClientException(
@@ -123,4 +120,44 @@ class Atomic extends Pipeline

        return $responses;
    }
+
+    /**
+     * @param  ConnectionInterface $connection
+     * @param  SplQueue            $commands
+     * @return void
+     * @throws Throwable
+     */
+    protected function queuePipeline(ConnectionInterface $connection, SplQueue $commands)
+    {
+        $commandFactory = $this->getClient()->getCommandFactory();
+        $this->writeToSingleNode($connection, $commands);
+
+        foreach ($commands as $command) {
+            $response = $connection->readResponse($command);
+
+            if ($response instanceof ErrorResponseInterface) {
+                $this->executeCommandWithRetry($connection, $commandFactory->create('discard'));
+                throw new ServerException($response->getMessage());
+            }
+        }
+    }
+
+    /**
+     * @param  ConnectionInterface $connection
+     * @param  Command             $command
+     * @return mixed
+     * @throws Throwable
+     */
+    protected function executeCommandWithRetry(ConnectionInterface $connection, CommandInterface $command)
+    {
+        $retry = $connection->getParameters()->retry;
+
+        return $retry->callWithRetry(function () use ($connection, $command) {
+            return $connection->executeCommand($command);
+        }, function (Throwable $e) {
+            if ($e instanceof CommunicationException) {
+                $e->getConnection()->disconnect();
+            }
+        });
+    }
 }
@@ -12,9 +12,11 @@

 namespace Predis\Pipeline;

+use Predis\CommunicationException;
 use Predis\Connection\AggregateConnectionInterface;
 use Predis\Connection\ConnectionInterface;
 use SplQueue;
+use Throwable;

 /**
 * Command pipeline that writes commands to the servers but discards responses.
@@ -26,11 +28,19 @@ class FireAndForget extends Pipeline
     */
    protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
    {
-        if ($connection instanceof AggregateConnectionInterface) {
-            $this->writeToMultiNode($connection, $commands);
-        } else {
-            $this->writeToSingleNode($connection, $commands);
-        }
+        $retry = $connection->getParameters()->retry;
+
+        $retry->callWithRetry(function () use ($connection, $commands) {
+            if ($connection instanceof AggregateConnectionInterface) {
+                $this->writeToMultiNode($connection, $commands);
+            } else {
+                $this->writeToSingleNode($connection, $commands);
+            }
+        }, function (Throwable $e) {
+            if ($e instanceof CommunicationException) {
+                $e->getConnection()->disconnect();
+            }
+        });

        $connection->disconnect();

@@ -18,13 +18,18 @@ use Predis\ClientContextInterface;
 use Predis\ClientException;
 use Predis\ClientInterface;
 use Predis\Command\CommandInterface;
+use Predis\CommunicationException;
 use Predis\Connection\AggregateConnectionInterface;
+use Predis\Connection\Cluster\RedisCluster;
+use Predis\Connection\ConnectionException;
 use Predis\Connection\ConnectionInterface;
 use Predis\Connection\Replication\ReplicationInterface;
 use Predis\Response\ErrorInterface as ErrorResponseInterface;
 use Predis\Response\ResponseInterface;
 use Predis\Response\ServerException;
+use Predis\TimeoutException;
 use SplQueue;
+use Throwable;

 /**
 * Implementation of a command pipeline in which write and read operations of
@@ -129,22 +134,65 @@ class Pipeline implements ClientContextInterface
     * @param SplQueue            $commands   Queued commands.
     *
     * @return array
+     * @throws Throwable
     */
    protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
    {
+        $retry = $connection->getParameters()->retry;
+        $backupQueue = $this->createDeepCloneQueue($commands);
+
+        return $retry->callWithRetry(
+            function () use ($connection, &$commands) {
+                return $this->executePipelineInternal($connection, $commands);
+            },
+            function (Throwable $e) use (&$commands, $backupQueue, $connection) {
+                if (!$e instanceof CommunicationException) {
+                    throw $e;
+                }
+
+                if ($connection instanceof AggregateConnectionInterface) {
+                    $this->onAggregateConnectionFailCallback($connection, $e);
+                } else {
+                    $connection = $e->getConnection();
+                    $connection->disconnect();
+                }
+
+                // In case of error whole pipeline should be retried
+                // So we need to write all original commands again
+                $commands = $this->createDeepCloneQueue($backupQueue);
+            }
+        );
+    }
+
+    /**
+     * @param  ConnectionInterface $connection
+     * @param  SplQueue            $commands
+     * @return array
+     * @throws ServerException
+     * @throws Throwable
+     */
+    protected function executePipelineInternal(
+        ConnectionInterface $connection,
+        SplQueue $commands
+    ): array {
+        $responses = [];
+        $exceptions = $this->throwServerExceptions();
+        $protocolVersion = (int) $connection->getParameters()->protocol;
+
        if ($connection instanceof AggregateConnectionInterface) {
            $this->writeToMultiNode($connection, $commands);
        } else {
            $this->writeToSingleNode($connection, $commands);
        }

-        $responses = [];
-        $exceptions = $this->throwServerExceptions();
-        $protocolVersion = (int) $connection->getParameters()->protocol;
-
        while (!$commands->isEmpty()) {
            $command = $commands->dequeue();
-            $response = $connection->readResponse($command);
+
+            if ($connection instanceof AggregateConnectionInterface) {
+                $response = $connection->getConnectionByCommand($command)->readResponse($command);
+            } else {
+                $response = $connection->readResponse($command);
+            }

            if (!$response instanceof ResponseInterface) {
                if ($protocolVersion === 2) {
@@ -162,12 +210,30 @@ class Pipeline implements ClientContextInterface
        return $responses;
    }

+    /**
+     * Creates a deep copy of commands queue for backup.
+     *
+     * @param  SplQueue $queue
+     * @return SplQueue
+     */
+    private function createDeepCloneQueue(SplQueue $queue): SplQueue
+    {
+        $new = new SplQueue();
+
+        foreach ($queue as $command) {
+            $new->enqueue(clone $command);
+        }
+
+        return $new;
+    }
+
    /**
     * Writes pipelined commands to single node connection.
     *
     * @param  ConnectionInterface $connection
     * @param  SplQueue            $commands
     * @return void
+     * @throws Throwable
     */
    protected function writeToSingleNode(ConnectionInterface $connection, SplQueue $commands)
    {
@@ -186,9 +252,12 @@ class Pipeline implements ClientContextInterface
     * @param  AggregateConnectionInterface $connection
     * @param  SplQueue                     $commands
     * @return void
+     * @throws Throwable
     */
    protected function writeToMultiNode(AggregateConnectionInterface $connection, SplQueue $commands)
    {
+        $retry = $connection->getParameters()->retry;
+
        foreach ($commands as $command) {
            $nodeConnection = $connection->getConnectionByCommand($command);
            $nodeConnection->write($command->serializeCommand());
@@ -286,4 +355,37 @@ class Pipeline implements ClientContextInterface
    {
        return $this->client;
    }
+
+    /**
+     * Handle aggregate connection exception.
+     *
+     * @param  AggregateConnectionInterface $connection
+     * @param  CommunicationException       $e
+     * @return void
+     */
+    private function onAggregateConnectionFailCallback(AggregateConnectionInterface $connection, Throwable $e)
+    {
+        if ($e instanceof ConnectionException) {
+            $nodeConnection = $e->getConnection();
+
+            if ($nodeConnection) {
+                $nodeConnection->disconnect();
+                $connection->remove($nodeConnection);
+            }
+
+            if ($connection instanceof RedisCluster) {
+                if ($connection->useClusterSlots) {
+                    $connection->askSlotMap();
+                }
+            }
+        }
+
+        if ($e instanceof TimeoutException) {
+            $nodeConnection = $e->getConnection();
+
+            if ($nodeConnection) {
+                $nodeConnection->disconnect();
+            }
+        }
+    }
 }
@@ -0,0 +1,143 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry;
+
+use Predis\Connection\ConnectionException;
+use Predis\Connection\Resource\Exception\StreamInitException;
+use Predis\Retry\Strategy\RetryStrategyInterface;
+use Predis\TimeoutException;
+use Throwable;
+
+class Retry
+{
+    /**
+     * @var RetryStrategyInterface
+     */
+    protected $backoffStrategy;
+
+    /**
+     * @var int
+     */
+    protected $retries;
+
+    /**
+     * @var array
+     */
+    protected $catchableExceptions = [
+        TimeoutException::class,
+        ConnectionException::class,
+        StreamInitException::class,
+    ];
+
+    /**
+     * @param RetryStrategyInterface $backoffStrategy
+     * @param int                    $retries
+     * @param array|null             $catchableExceptions A list of exceptions classes that should be caught.
+     *                                                    Overrides default list of the catchable exceptions.
+     */
+    public function __construct(
+        RetryStrategyInterface $backoffStrategy,
+        int $retries,
+        ?array $catchableExceptions = null
+    ) {
+        $this->backoffStrategy = $backoffStrategy;
+        $this->retries = $retries;
+
+        if (null !== $catchableExceptions) {
+            $this->catchableExceptions = $catchableExceptions;
+        }
+    }
+
+    /**
+     * Update the retry count.
+     *
+     * @param  int  $retries
+     * @return void
+     */
+    public function updateRetriesCount(int $retries): void
+    {
+        $this->retries = $retries;
+    }
+
+    /**
+     * Extend catchable exceptions list.
+     *
+     * @param  array $catchableExceptions
+     * @return void
+     */
+    public function updateCatchableExceptions(array $catchableExceptions): void
+    {
+        $this->catchableExceptions = array_merge($this->catchableExceptions, $catchableExceptions);
+    }
+
+    /**
+     * @return int
+     */
+    public function getRetries(): int
+    {
+        return $this->retries;
+    }
+
+    /**
+     * @return RetryStrategyInterface
+     */
+    public function getStrategy(): RetryStrategyInterface
+    {
+        return $this->backoffStrategy;
+    }
+
+    /**
+     * @param  callable(): mixed              $do
+     * @param  callable(Throwable): void|null $fail
+     * @return mixed
+     * @throws Throwable
+     */
+    public function callWithRetry(callable $do, ?callable $fail = null)
+    {
+        $failures = 0;
+
+        while (true) {
+            try {
+                return $do();
+            } catch (Throwable $e) {
+                if (null !== $this->catchableExceptions) {
+                    $isCatchable = false;
+                    foreach ($this->catchableExceptions as $catchableException) {
+                        if ($e instanceof $catchableException) {
+                            $isCatchable = true;
+                        }
+                    }
+
+                    if (!$isCatchable) {
+                        throw $e;
+                    }
+                }
+
+                $backoff = $this->backoffStrategy->compute($failures);
+                ++$failures;
+
+                if ($this->retries >= 0 && $failures > $this->retries) {
+                    throw $e;
+                }
+
+                if ($fail !== null) {
+                    $fail($e);
+                }
+
+                if ($backoff > 0) {
+                    usleep($backoff);
+                }
+            }
+        }
+    }
+}
@@ -0,0 +1,37 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+/**
+ * Equal backoff between retry.
+ */
+class EqualBackoff implements RetryStrategyInterface
+{
+    /**
+     * @var int
+     */
+    protected $backoff;
+
+    /**
+     * @param int $backoff in micro seconds
+     */
+    public function __construct(int $backoff)
+    {
+        $this->backoff = $backoff;
+    }
+
+    public function compute(int $failures): int
+    {
+        return $this->backoff;
+    }
+}
@@ -0,0 +1,75 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+class ExponentialBackoff implements RetryStrategyInterface
+{
+    /**
+     * @var int
+     */
+    protected $base;
+
+    /**
+     * @var int
+     */
+    protected $cap;
+
+    /**
+     * @var bool
+     */
+    protected $withJitter;
+
+    /**
+     * @param int  $base       in micro seconds
+     * @param int  $cap        in micro seconds
+     * @param bool $withJitter
+     */
+    public function __construct(int $base = self::DEFAULT_BASE, int $cap = self::DEFAULT_CAP, bool $withJitter = false)
+    {
+        $this->base = $base;
+        $this->cap = $cap;
+        $this->withJitter = $withJitter;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function compute(int $failures): int
+    {
+        if ($this->withJitter) {
+            return min($this->cap, (mt_rand(0, mt_getrandmax() - 1) / mt_getrandmax()) * ($this->base * 2 ** $failures));
+        }
+
+        if ($this->cap > 0) {
+            return min($this->cap, $this->base * 2 ** $failures);
+        }
+
+        return $this->base * 2 ** $failures;
+    }
+
+    /**
+     * @return int
+     */
+    public function getBase(): int
+    {
+        return $this->base;
+    }
+
+    /**
+     * @return int
+     */
+    public function getCap(): int
+    {
+        return $this->cap;
+    }
+}
@@ -0,0 +1,24 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+/**
+ * No backoff between retry.
+ */
+class NoBackoff extends EqualBackoff
+{
+    public function __construct()
+    {
+        parent::__construct(0);
+    }
+}
@@ -0,0 +1,37 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+/**
+ * Retry strategy interface.
+ */
+interface RetryStrategyInterface
+{
+    /**
+     * Minimum backoff between each retry in micro seconds.
+     */
+    public const DEFAULT_BASE = 8 * 1000;
+
+    /**
+     * Maximum backoff between each retry in micro seconds.
+     */
+    public const DEFAULT_CAP = 512 * 1000;
+
+    /**
+     * Compute backoff in micro seconds upon failure.
+     *
+     * @param  int $failures
+     * @return int
+     */
+    public function compute(int $failures): int;
+}
@@ -0,0 +1,24 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis;
+
+use Predis\Connection\NodeConnectionInterface;
+use Throwable;
+
+class TimeoutException extends CommunicationException
+{
+    public function __construct(NodeConnectionInterface $connection, $code = 0, ?Throwable $previous = null)
+    {
+        parent::__construct($connection, 'Operation has timed out', $code, $previous);
+    }
+}
@@ -18,13 +18,17 @@ use Predis\Command\Redis\EXEC;
 use Predis\Command\Redis\MULTI;
 use Predis\Command\Redis\UNWATCH;
 use Predis\Command\Redis\WATCH;
+use Predis\CommunicationException;
+use Predis\Connection\ConnectionException;
 use Predis\Connection\NodeConnectionInterface;
 use Predis\Connection\RelayConnection;
 use Predis\Connection\Replication\ReplicationInterface;
 use Predis\Response\ErrorInterface;
 use Predis\Response\ServerException;
+use Predis\TimeoutException;
 use Predis\Transaction\MultiExecState;
 use Predis\Transaction\Response\BypassTransactionResponse;
+use Throwable;

 /**
 * Defines strategy for connections that operates on non-distributed hash slots.
@@ -60,6 +64,7 @@ abstract class NonClusterConnectionStrategy implements StrategyInterface

    /**
     * {@inheritDoc}
+     * @throws Throwable
     */
    public function executeCommand(CommandInterface $command)
    {
@@ -67,7 +72,15 @@ abstract class NonClusterConnectionStrategy implements StrategyInterface
            return $this->executeBypassingTransaction($command);
        }

-        return $this->connection->executeCommand($command);
+        $retry = $this->connection->getParameters()->retry;
+
+        return $retry->callWithRetry(
+            function () use ($command) {
+                return $this->connection->executeCommand($command);
+            }, function (CommunicationException $e) {
+                $this->onFailCallback($e);
+            }
+        );
    }

    /**
@@ -99,10 +112,19 @@ abstract class NonClusterConnectionStrategy implements StrategyInterface

    /**
     * {@inheritDoc}
+     * @throws Throwable
     */
    public function unwatch()
    {
-        return $this->connection->executeCommand(new UNWATCH());
+        $retry = $this->connection->getParameters()->retry;
+
+        return $retry->callWithRetry(
+            function () {
+                return $this->connection->executeCommand(new UNWATCH());
+            }, function (CommunicationException $e) {
+                $this->onFailCallback($e);
+            }
+        );
    }

    /**
@@ -118,12 +140,20 @@ abstract class NonClusterConnectionStrategy implements StrategyInterface
     *
     * @param  CommandInterface          $command
     * @return BypassTransactionResponse
-     * @throws ServerException
+     * @throws ServerException|Throwable
     */
    protected function executeBypassingTransaction(CommandInterface $command): BypassTransactionResponse
    {
+        $retry = $this->connection->getParameters()->retry;
+
        try {
-            $response = $this->connection->executeCommand($command);
+            $response = $retry->callWithRetry(
+                function () use ($command) {
+                    return $this->connection->executeCommand($command);
+                }, function (CommunicationException $e) {
+                    $this->onFailCallback($e);
+                }
+            );
        } catch (ServerException $exception) {
            if (!$this->connection instanceof RelayConnection) {
                throw $exception;
@@ -146,4 +176,38 @@ abstract class NonClusterConnectionStrategy implements StrategyInterface

        return new BypassTransactionResponse($response);
    }
+
+    /**
+     * Handle communication exception.
+     *
+     * @param  CommunicationException $e
+     * @return void
+     */
+    private function onFailCallback(CommunicationException $e)
+    {
+        $connection = $e->getConnection();
+
+        if ($connection instanceof NodeConnectionInterface) {
+            $connection->disconnect();
+
+            return;
+        }
+
+        if ($e instanceof ConnectionException) {
+            $nodeConnection = $e->getConnection();
+
+            if ($nodeConnection) {
+                $nodeConnection->disconnect();
+                $this->connection->remove($nodeConnection);
+            }
+        }
+
+        if ($e instanceof TimeoutException) {
+            $nodeConnection = $e->getConnection();
+
+            if ($nodeConnection) {
+                $nodeConnection->disconnect();
+            }
+        }
+    }
 }
@@ -15,6 +15,7 @@ namespace Predis\Connection;
 use PHPUnit\Framework\MockObject\MockObject;
 use Predis\Command\CommandInterface;
 use Predis\Command\RawCommand;
+use Predis\TimeoutException;
 use PredisTestCase;

 /**
@@ -340,6 +341,7 @@ abstract class PredisConnectionTestCase extends PredisTestCase

    /**
     * @group connected
+     * @requiresRedisVersion >= 6.0.0
     */
    public function testSendsInitializationCommandsOnConnection(): void
    {
@@ -459,7 +461,7 @@ abstract class PredisConnectionTestCase extends PredisTestCase
     */
    public function testThrowsExceptionOnReadWriteTimeout(): void
    {
-        $this->expectException('Predis\Connection\ConnectionException');
+        $this->expectException(TimeoutException::class);

        $commands = $this->getCommandFactory();

@@ -180,12 +180,14 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
        } elseif ($this->isUnprotectedTest()) {
            $port = constant('REDIS_UNPROTECTED_SERVER_PORT');
            $password = '';
+        } elseif ($this->isSSLTest()) {
+            $port = getenv('REDIS_SSL_PORT');
        } else {
            $port = constant('REDIS_SERVER_PORT');
        }

        return [
-            'scheme' => 'tcp',
+            'scheme' => $this->isSSLTest() ? 'tls' : 'tcp',
            'host' => constant('REDIS_SERVER_HOST'),
            'port' => $port,
            'database' => constant('REDIS_SERVER_DBNUM'),
@@ -256,7 +258,7 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
     *
     * @return Client
     */
-    protected function createClient(?array $parameters = null, ?array $options = null, ?bool $flushdb = true): Client
+    public function createClient(?array $parameters = null, ?array $options = null, ?bool $flushdb = true): Client
    {
        $parameters = array_merge(
            $this->getDefaultParametersArray(),
@@ -282,6 +284,28 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
                ],
                $options
            );
+
+            if ($this->isSSLTest()) {
+                $options = array_merge($options, [
+                    'parameters' => [
+                        'ssl' => [
+                            'cafile' => getenv('CLUSTER_CA_CERT_PATH'),
+                            'verify_peer' => true,
+                            'verify_peer_name' => false,
+                        ],
+                    ],
+                ]);
+            }
+        } else {
+            if ($this->isSSLTest()) {
+                $parameters = array_merge($parameters, [
+                    'ssl' => [
+                        'cafile' => getenv('STANDALONE_CA_CERT_PATH'),
+                        'verify_peer' => true,
+                        'verify_peer_name' => false,
+                    ],
+                ]);
+            }
        }

        $client = new Client($parameters, $options);
@@ -294,6 +318,54 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
        return $client;
    }

+    /**
+     * Creates a client for version checking without SSL configuration.
+     * This is used to check Redis version before attempting SSL connection.
+     *
+     * @return Client
+     */
+    protected function createClientForVersionCheck(): Client
+    {
+        // For SSL tests, temporarily override to use non-SSL configuration
+        $isSSL = $this->isSSLTest();
+        $isCluster = $this->isClusterTest();
+
+        if ($isSSL && $isCluster) {
+            // For cluster SSL tests, use non-SSL cluster endpoints
+            $endpoints = explode(',', constant('REDIS_CLUSTER_ENDPOINTS'));
+            $parameters = array_map(static function (string $elem) {
+                return 'tcp://' . $elem;
+            }, $endpoints);
+        } elseif ($isSSL) {
+            // For standalone SSL tests, use non-SSL port
+            $parameters = [
+                'scheme' => 'tcp',
+                'host' => constant('REDIS_SERVER_HOST'),
+                'port' => constant('REDIS_SERVER_PORT'),
+                'database' => constant('REDIS_SERVER_DBNUM'),
+                'password' => getenv('REDIS_PASSWORD') ?: constant('REDIS_PASSWORD'),
+            ];
+        } else {
+            // For non-SSL tests, use default parameters
+            $parameters = $this->getDefaultParametersArray();
+        }
+
+        $commandsFactory = $this->getCommandFactory();
+        $options = array_merge(
+            ['commands' => $commandsFactory],
+            getenv('USE_RELAY') ? ['connections' => 'relay'] : []
+        );
+
+        if ($isCluster) {
+            $options['cluster'] = 'redis';
+        }
+
+        $client = new Client($parameters, $options);
+        $client->connect();
+
+        return $client;
+    }
+
    /**
     * Returns a basic mock object of a connection to a single Redis node.
     *
@@ -371,7 +443,9 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
        if (isset($this->info)) {
            $info = $this->info;
        } else {
-            $client = $this->createClient(null, null, true);
+            // For SSL tests, connect to non-SSL port to check version first
+            // This prevents connection failures on Redis < 7.2.0 which doesn't support SSL
+            $client = $this->createClientForVersionCheck();
            $info = array_change_key_case($client->info());
            $this->info = $info;
        }
@@ -383,7 +457,7 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
            // Redis < 2.6
            $version = $info['redis_version'];
        } else {
-            $client = $this->createClient(null, null, true);
+            $client = $this->createClientForVersionCheck();
            $connection = $client->getConnection();
            throw new RuntimeException("Unable to retrieve a valid server info payload from $connection");
        }
@@ -607,6 +681,34 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
            && in_array('cluster', $annotations['method']['group'], true);
    }

+    /**
+     * Check annotations if it's matches to SSL test scenario.
+     *
+     * @return bool
+     */
+    protected function isSSLTest(): bool
+    {
+        $annotations = TestUtil::parseTestMethodAnnotations(
+            get_class($this),
+            $this->getName(false)
+        );
+
+        $annotationExists = isset($annotations['method']['requiresRedisVersion']);
+
+        if (!$annotationExists) {
+            foreach ($this->modulesMapping as $module => $configuration) {
+                if (isset($annotations['method'][$configuration['annotation']])) {
+                    $annotationExists = true;
+                }
+            }
+        }
+
+        return $annotationExists
+            && isset($annotations['method']['group'])
+            && in_array('connected', $annotations['method']['group'], true)
+            && in_array('ssl', $annotations['method']['group'], true);
+    }
+
    /**
     * Check annotations if it's matches to stack test scenario.
     *
@@ -646,10 +748,14 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase
     */
    protected function prepareClusterEndpoints(): array
    {
-        $endpoints = explode(',', constant('REDIS_CLUSTER_ENDPOINTS'));
+        $endpoints = explode(
+            ',',
+            constant($this->isSSLTest() ? 'SSL_REDIS_CLUSTER_ENDPOINTS' : 'REDIS_CLUSTER_ENDPOINTS')
+        );
+        $scheme = $this->isSSLTest() ? 'tls' : 'tcp';

-        return array_map(static function (string $elem) {
-            return 'tcp://' . $elem;
+        return array_map(static function (string $elem) use ($scheme) {
+            return "{$scheme}://" . $elem;
        }, $endpoints);
    }
 }
@@ -12,16 +12,26 @@

 namespace Predis;

+use Exception;
 use Iterator;
 use PHPUnit\Framework\MockObject\MockObject;
 use Predis\Command\Factory as CommandFactory;
 use Predis\Command\Processor\KeyPrefixProcessor;
+use Predis\Command\RawCommand;
+use Predis\Connection\Cluster\RedisCluster;
+use Predis\Connection\Factory;
 use Predis\Connection\NodeConnectionInterface;
 use Predis\Connection\Parameters;
 use Predis\Connection\ParametersInterface;
 use Predis\Connection\Replication\MasterSlaveReplication;
+use Predis\Connection\Resource\StreamFactoryInterface;
+use Predis\Connection\StreamConnection;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
 use PredisTestCase;
+use Psr\Http\Message\StreamInterface;
 use ReflectionProperty;
+use RuntimeException;
 use stdClass;

 class ClientTest extends PredisTestCase
@@ -188,7 +198,7 @@ class ClientTest extends PredisTestCase
     */
    public function testConstructorWithConnectionArgument(): void
    {
-        $factory = new Connection\Factory();
+        $factory = new Factory();
        $connection = $factory->create('tcp://localhost:7000');

        $client = new Client($connection);
@@ -211,7 +221,7 @@ class ClientTest extends PredisTestCase
    {
        $cluster = new Connection\Cluster\PredisCluster(new Parameters());

-        $factory = new Connection\Factory();
+        $factory = new Factory();
        $cluster->add($factory->create('tcp://localhost:7000'));
        $cluster->add($factory->create('tcp://localhost:7001'));

@@ -228,7 +238,7 @@ class ClientTest extends PredisTestCase
    {
        $replication = new MasterSlaveReplication();

-        $factory = new Connection\Factory();
+        $factory = new Factory();
        $replication->add($factory->create('tcp://host1?alias=master'));
        $replication->add($factory->create('tcp://host2?alias=slave'));

@@ -610,6 +620,10 @@ class ClientTest extends PredisTestCase
            ->expects($this->once())
            ->method('executeCommand')
            ->willReturn($expectedResponse);
+        $connection
+            ->expects($this->once())
+            ->method('getParameters')
+            ->willReturn(new Parameters());

        $client = new Client($connection);
        $client->executeCommand($ping);
@@ -628,6 +642,10 @@ class ClientTest extends PredisTestCase
            ->expects($this->once())
            ->method('executeCommand')
            ->willReturn($expectedResponse);
+        $connection
+            ->expects($this->once())
+            ->method('getParameters')
+            ->willReturn(new Parameters());

        $client = new Client($connection, ['exceptions' => false]);
        $response = $client->executeCommand($ping);
@@ -689,6 +707,11 @@ class ClientTest extends PredisTestCase
            ->with($this->isRedisCommand('PING'))
            ->willReturn($expectedResponse);

+        $connection
+            ->expects($this->once())
+            ->method('getParameters')
+            ->willReturn(new Parameters());
+
        $client = new Client($connection);
        $client->ping();
    }
@@ -706,6 +729,10 @@ class ClientTest extends PredisTestCase
            ->method('executeCommand')
            ->with($this->isRedisCommand('PING'))
            ->willReturn($expectedResponse);
+        $connection
+            ->expects($this->once())
+            ->method('getParameters')
+            ->willReturn(new Parameters());

        $client = new Client($connection, ['exceptions' => false]);
        $response = $client->ping();
@@ -957,7 +984,7 @@ class ClientTest extends PredisTestCase
     */
    public function testGetClientByMethodSupportsSelectingConnectionByCommand(): void
    {
-        $command = Command\RawCommand::create('GET', 'key');
+        $command = RawCommand::create('GET', 'key');
        $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock();

        $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface')
@@ -1185,6 +1212,10 @@ class ClientTest extends PredisTestCase
            ->expects($this->once())
            ->method('executeCommand')
            ->willReturn(new Response\Status('QUEUED'));
+        $connection
+            ->expects($this->any())
+            ->method('getParameters')
+            ->willReturn(new Parameters());

        $callable = $this->getMockBuilder('stdClass')
            ->addMethods(['__invoke'])
@@ -1308,6 +1339,50 @@ class ClientTest extends PredisTestCase
        $this->assertSame('127.0.0.1:6381', $iterator->key());
    }

+    /**
+     * @group disconnected
+     */
+    public function testExecuteCommandRetryCommandOnRetryableException()
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $mockStream
+            ->expects($this->atLeast(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(4))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000
+            );
+
+        $mockStream
+            ->expects($this->once())
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+PONG\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection = new StreamConnection($parameters, $mockStreamFactory);
+        $client = new Client($connection);
+        $this->assertEquals('PONG', $client->ping());
+    }
+
    /**
     * @group connected
     * @group relay-incompatible
@@ -1439,6 +1514,172 @@ class ClientTest extends PredisTestCase
        $this->assertEquals(1, $clientTestUser->acl->delUser('test_user'));
    }

+    /**
+     * @group connected
+     * @return void
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testStandaloneNodeRetryCommandExecutionOnTimeoutException(): void
+    {
+        $retries = 0;
+        $mockDisconnect = function () use (&$retries) {
+            $streamConnection = new StreamConnection(new Parameters([
+                'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+            ]));
+            $disconnectFunc = [$streamConnection, 'disconnect'];
+            ++$retries;
+            $disconnectFunc();
+        };
+
+        $stubConnection = $this->getMockBuilder(StreamConnection::class)
+            ->setConstructorArgs([new Parameters([
+                'retry' => new Retry(new ExponentialBackoff(100, 1000), 3),
+                'read_write_timeout' => 0.1,
+            ])])
+            ->onlyMethods(['disconnect'])
+            ->getMock();
+
+        $stubConnection
+            ->expects($this->exactly(7))
+            ->method('disconnect')
+            ->willReturnCallback($mockDisconnect);
+
+        $stubConnection->addConnectCommand(new RawCommand('auth', ['foobar']));
+
+        $client = new Client($stubConnection);
+
+        $this->expectException(TimeoutException::class);
+
+        $client->blmpop(3, ['random_key']);
+        $this->assertEquals(3, $retries);
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @return void
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testStandaloneNodeRetryCommandExecutionOnTimeoutExceptionIntegration(): void
+    {
+        // Retry used to wrap callback around, so we can count retries
+        $retry = new Retry(new ExponentialBackoff(100, 1000), 3);
+        $retriesCount = 0;
+        $retryWrapperFunc = function (callable $do, ?callable $fail = null) use ($retry, &$retriesCount) {
+            $failWrapperFunc = function (Exception $e) use (&$retriesCount, $fail) {
+                ++$retriesCount;
+                $fail($e);
+            };
+
+            return $retry->callWithRetry($do, $failWrapperFunc);
+        };
+
+        $mockRetry = $this->getMockBuilder(Retry::class)
+            ->setConstructorArgs([new ExponentialBackoff(100, 1000), 3])
+            ->onlyMethods(['callWithRetry'])
+            ->getMock();
+
+        $mockRetry
+            ->expects($this->any())
+            ->method('callWithRetry')
+            ->willReturnCallback($retryWrapperFunc);
+
+        // Create a real connection with mocked retry and short read_write_timeout
+        $client = $this->createClient([
+            'retry' => $mockRetry,
+            'read_write_timeout' => 0.1,
+        ]);
+
+        $this->expectException(TimeoutException::class);
+
+        try {
+            // blmpop with 3 second timeout will exceed the 0.1 second read_write_timeout
+            // causing TimeoutException to be thrown and retried 3 times before failing
+            $client->blmpop(3, ['random_key_that_does_not_exist']);
+        } finally {
+            $this->assertGreaterThanOrEqual(3, $retriesCount);
+        }
+    }
+
+    /**
+     * @group connected
+     * @group cluster
+     * @return void
+     * @requiresRedisVersion >= 2.0.0
+     */
+    public function testClusterRetryCommandExecutionOnTimeoutException(): void
+    {
+        $defaultParams = $this->getDefaultParametersArray();
+        $parsedParams = [];
+
+        foreach ($defaultParams as $param) {
+            $parsedParam = Parameters::parse($param);
+            $parsedParam['retry'] = new Retry(new ExponentialBackoff(1000, 10000), 3);
+            $parsedParam['read_write_timeout'] = 0.1;
+            $parsedParams[] = Parameters::create($parsedParam);
+        }
+
+        $retries = 0;
+        $mockDisconnect = function () use (&$retries) {
+            $streamConnection = new StreamConnection(new Parameters([
+                'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+            ]));
+            $disconnectFunc = [$streamConnection, 'disconnect'];
+            ++$retries;
+            $disconnectFunc();
+        };
+
+        $stubConnection1 = $this->getMockBuilder(StreamConnection::class)
+            ->setConstructorArgs([$parsedParams[0]])
+            ->onlyMethods(['disconnect'])
+            ->getMock();
+
+        $stubConnection1
+            ->expects($this->any())
+            ->method('disconnect')
+            ->willReturnCallback($mockDisconnect);
+
+        $stubConnection1->addConnectCommand(new RawCommand('auth', [$parsedParams[0]->password]));
+
+        $stubConnection2 = $this->getMockBuilder(StreamConnection::class)
+            ->setConstructorArgs([$parsedParams[1]])
+            ->onlyMethods(['disconnect'])
+            ->getMock();
+
+        $stubConnection2
+            ->expects($this->any())
+            ->method('disconnect')
+            ->willReturnCallback($mockDisconnect);
+
+        $stubConnection2->addConnectCommand(new RawCommand('auth', [$parsedParams[1]->password]));
+
+        $stubConnection3 = $this->getMockBuilder(StreamConnection::class)
+            ->setConstructorArgs([$parsedParams[2]])
+            ->onlyMethods(['disconnect'])
+            ->getMock();
+
+        $stubConnection3
+            ->expects($this->any())
+            ->method('disconnect')
+            ->willReturnCallback($mockDisconnect);
+
+        $stubConnection3->addConnectCommand(new RawCommand('auth', [$parsedParams[2]->password]));
+
+        $mockFactory = $this->getMockBuilder(Factory::class)->getMock();
+        $clusterConnection = new RedisCluster($mockFactory, $parsedParams[0]);
+
+        $clusterConnection->add($stubConnection1);
+        $clusterConnection->add($stubConnection2);
+        $clusterConnection->add($stubConnection3);
+
+        $client = new Client($clusterConnection);
+
+        $this->expectException(TimeoutException::class);
+
+        $client->blpop(['random_key'], 3);
+        $this->assertEquals(3, $retries);
+    }
+
    // ******************************************************************** //
    // ---- HELPER METHODS ------------------------------------------------ //
    // ******************************************************************** //
@@ -457,6 +457,10 @@ class KeyPrefixProcessorTest extends PredisTestCase
                ['key', 1.0, 'member'],
                ['prefix:key', 1.0, 'member'],
            ],
+            ['ZRANDMEMBER',
+                ['key', 10],
+                ['prefix:key', 10],
+            ],
            ['ZREM',
                ['key', 'member1', 'member2', 'member3'],
                ['prefix:key', 'member1', 'member2', 'member3'],
@@ -53,8 +53,6 @@ class VINFO_Test extends PredisCommandTestCase

    /**
     * @group connected
-     * @group relay-incompatible
-     * @group relay-fixme
     * @return void
     * @requiresRedisVersion >= 8.0.0
     */
@@ -76,8 +74,6 @@ class VINFO_Test extends PredisCommandTestCase

    /**
     * @group connected
-     * @group relay-incompatible
-     * @group relay-fixme
     * @return void
     * @requiresRedisVersion >= 8.0.0
     */
@@ -0,0 +1,120 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Command\Redis;
+
+use Predis\Command\PrefixableCommand;
+use Predis\Command\Redis\Utils\VectorUtility;
+
+class VRANGE_Test extends PredisCommandTestCase
+{
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return VRANGE::class;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'VRANGE';
+    }
+
+    /**
+     * @return void
+     */
+    public function testFilterArguments(): void
+    {
+        $command = $this->getCommand();
+        $command->setArguments(['key', 'start', 'end']);
+
+        $this->assertSame(['key', 'start', 'end'], $command->getArguments());
+
+        $command->setArguments(['key', 'start', 'end', 3]);
+        $this->assertSame(['key', 'start', 'end', 3], $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testPrefixKeys(): void
+    {
+        /** @var PrefixableCommand $command */
+        $command = $this->getCommand();
+        $actualArguments = ['key', '-', '+'];
+        $prefix = 'prefix:';
+        $expectedArguments = ['prefix:key', '-', '+'];
+
+        $command->setArguments($actualArguments);
+        $command->prefixKeys($prefix);
+
+        $this->assertSame($expectedArguments, $command->getArguments());
+    }
+
+    /**
+     * @return void
+     */
+    public function testParseResponse(): void
+    {
+        $this->assertEquals(1, $this->getCommand()->parseResponse(1));
+    }
+
+    /**
+     * @group connected
+     * @return void
+     * @requiresRedisVersion >= 8.4.0
+     */
+    public function testReturnsRandomMember(): void
+    {
+        $redis = $this->getClient();
+
+        $this->assertTrue(
+            $redis->vadd('key', VectorUtility::toBlob([0.1, 0.2, 0.3, 0.4]), 'elem1', 10)
+        );
+        $this->assertTrue(
+            $redis->vadd('key', [0.1, 0.2, 0.3, 0.4], 'elem2', 10)
+        );
+        $this->assertTrue(
+            $redis->vadd('key', [0.1, 0.2, 0.3, 0.4], 'elem3', 10)
+        );
+
+        $this->assertSame(['elem1', 'elem2', 'elem3'], $redis->vrange('key', '-', '+'));
+        $this->assertSame(['elem1', 'elem2'], $redis->vrange('key', '-', '+', 2));
+    }
+
+    /**
+     * @group connected
+     * @return void
+     * @requiresRedisVersion >= 8.4.0
+     */
+    public function testReturnsRandomMemberResp3(): void
+    {
+        $redis = $this->getResp3Client();
+
+        $this->assertTrue(
+            $redis->vadd('key', VectorUtility::toBlob([0.1, 0.2, 0.3, 0.4]), 'elem1', 10)
+        );
+        $this->assertTrue(
+            $redis->vadd('key', [0.1, 0.2, 0.3, 0.4], 'elem2', 10)
+        );
+        $this->assertTrue(
+            $redis->vadd('key', [0.1, 0.2, 0.3, 0.4], 'elem3', 10)
+        );
+
+        $this->assertSame(['elem1', 'elem2', 'elem3'], $redis->vrange('key', '-', '+'));
+        $this->assertSame(['elem1', 'elem2'], $redis->vrange('key', '-', '+', 2));
+    }
+}
@@ -12,6 +12,7 @@

 namespace Predis\Command\Redis;

+use Predis\Command\PrefixableCommand;
 use Predis\Response\ServerException;

 class ZRANDMEMBER_test extends PredisCommandTestCase
@@ -93,6 +94,23 @@ class ZRANDMEMBER_test extends PredisCommandTestCase
        $this->assertNull($redis->zrandmember($notExpectedKey));
    }

+    /**
+     * @group disconnected
+     */
+    public function testPrefixKeys(): void
+    {
+        /** @var PrefixableCommand $command */
+        $command = $this->getCommand();
+        $actualArguments = ['key', 10];
+        $prefix = 'prefix:';
+        $expectedArguments = ['prefix:key', 10];
+
+        $command->setArguments($actualArguments);
+        $command->prefixKeys($prefix);
+
+        $this->assertSame($expectedArguments, $command->getArguments());
+    }
+
    /**
     * @group connected
     * @requiresRedisVersion >= 6.2.0
@@ -19,8 +19,14 @@ use Predis\Command;
 use Predis\Connection;
 use Predis\Connection\FactoryInterface;
 use Predis\Connection\Parameters;
+use Predis\Connection\Resource\StreamFactoryInterface;
+use Predis\Connection\StreamConnection;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
 use PredisTestCase;
+use Psr\Http\Message\StreamInterface;
+use RuntimeException;

 class RedisClusterTest extends PredisTestCase
 {
@@ -1373,6 +1379,59 @@ class RedisClusterTest extends PredisTestCase
        $cluster->executeCommand($command);
    }

+    /**
+     * @medium
+     * @group disconnected
+     * @group slow
+     */
+    public function testRetryCommandFailureOnCustomRetryConfiguration()
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $mockStream
+            ->expects($this->exactly(4))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(4))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000,
+                1000
+            );
+
+        $mockStream
+            ->expects($this->once())
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection = new StreamConnection($parameters, $mockStreamFactory);
+        $cluster = new RedisCluster(new Connection\Factory(), $parameters);
+        $cluster->useClusterSlots(false);
+        $cluster->add($connection);
+
+        $this->assertEquals(
+            'OK',
+            $cluster->executeCommand(Command\RawCommand::create('SET', 1001))
+        );
+    }
+
    /**
     * @medium
     * @group disconnected
@@ -12,6 +12,8 @@

 namespace Predis\Connection;

+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\NoBackoff;
 use PredisTestCase;

 class ParametersTest extends PredisTestCase
@@ -430,6 +432,7 @@ class ParametersTest extends PredisTestCase
            'host' => '127.0.0.1',
            'port' => 6379,
            'protocol' => 2,
+            'retry' => new Retry(new NoBackoff(), 0),
        ];
    }

@@ -15,9 +15,16 @@ namespace Predis\Connection\Replication;
 use PHPUnit\Framework\MockObject\MockObject;
 use Predis\Command;
 use Predis\Connection;
+use Predis\Connection\Parameters;
+use Predis\Connection\Resource\StreamFactoryInterface;
+use Predis\Connection\StreamConnection;
 use Predis\Replication\ReplicationStrategy;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
 use PredisTestCase;
+use Psr\Http\Message\StreamInterface;
+use RuntimeException;

 class MasterSlaveReplicationTest extends PredisTestCase
 {
@@ -1489,6 +1496,58 @@ repl_backlog_histlen:12978
        $replication->write($command1->serializeCommand() . $command2->serializeCommand() . $command3->serializeCommand());
    }

+    /**
+     * @medium
+     * @group disconnected
+     * @group slow
+     */
+    public function testRetryCommandFailureOnCustomRetryConfiguration()
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+            'role' => 'master',
+        ]);
+
+        $mockStream
+            ->expects($this->exactly(4))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(4))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000
+            );
+
+        $mockStream
+            ->expects($this->once())
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection = new StreamConnection($parameters, $mockStreamFactory);
+        $replication = new MasterSlaveReplication();
+        $replication->add($connection);
+
+        $this->assertEquals(
+            'OK',
+            $replication->executeCommand(Command\RawCommand::create('SET', 1001))
+        );
+    }
+
    public function connectionsProvider(): array
    {
        return [
@@ -16,10 +16,17 @@ use Exception;
 use PHPUnit\Framework\MockObject\MockObject;
 use Predis\Command;
 use Predis\Connection;
+use Predis\Connection\Parameters;
+use Predis\Connection\Resource\StreamFactoryInterface;
+use Predis\Connection\StreamConnection;
 use Predis\Replication;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
 use PredisTestCase;
+use Psr\Http\Message\StreamInterface;
 use ReflectionProperty;
+use RuntimeException;

 class SentinelReplicationTest extends PredisTestCase
 {
@@ -103,7 +110,7 @@ class SentinelReplicationTest extends PredisTestCase
     */
    public function testConnectionParametersInstanceForSentinelConnectionIsNotModified(): void
    {
-        $originalParameters = Connection\Parameters::create(
+        $originalParameters = Parameters::create(
            'tcp://127.0.0.1:5381?role=sentinel&database=1&password=secret'
        );

@@ -123,8 +130,8 @@ class SentinelReplicationTest extends PredisTestCase
     */
    public function testConnectionParametersInstanceForSentinelConnectionIsNotModifiedEmptyPassword(): void
    {
-        $sentinel1 = Connection\Parameters::create('tcp://127.0.0.1:5381?role=sentinel&database=1&password=');
-        $sentinel2 = Connection\Parameters::create('tcp://127.0.0.1:5381?role=sentinel&database=1');
+        $sentinel1 = Parameters::create('tcp://127.0.0.1:5381?role=sentinel&database=1&password=');
+        $sentinel2 = Parameters::create('tcp://127.0.0.1:5381?role=sentinel&database=1');

        $replication1 = $this->getReplicationConnection('svc', [$sentinel1]);
        $replication2 = $this->getReplicationConnection('svc', [$sentinel2]);
@@ -1774,6 +1781,73 @@ class SentinelReplicationTest extends PredisTestCase
        $this->assertSame($slave2, $replication->getCurrent());
    }

+    /**
+     * @medium
+     * @group disconnected
+     * @group slow
+     */
+    public function testRetryCommandFailureOnCustomRetryConfiguration()
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+            'role' => 'master',
+        ]);
+
+        $mockStream
+            ->expects($this->exactly(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                1000,
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000,
+                1000
+            );
+
+        $mockStream
+            ->expects($this->exactly(7))
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls("*1\r\n", "$5\r\n", "master\r\n", "*1\r\n", "$5\r\n", "master\r\n", "+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection = new StreamConnection($parameters, $mockStreamFactory);
+
+        $mockFactory = $this->getMockBuilder(Connection\Factory::class)->getMock();
+        $mockFactory
+            ->expects($this->any())
+            ->method('create')
+            ->willReturn($connection);
+
+        $sentinel = $this->getMockSentinelConnection();
+        $sentinel
+            ->expects($this->any())
+            ->method('executeCommand')
+            ->willReturn(['127.0.0.1', '6381']);
+
+        $replication = new SentinelReplication('src', [$sentinel], $mockFactory);
+        $replication->add($connection);
+
+        $this->assertEquals(
+            'OK',
+            $replication->executeCommand(Command\RawCommand::create('SET', 1001))
+        );
+    }
+
    // ******************************************************************** //
    // ---- HELPER METHODS ------------------------------------------------ //
    // ******************************************************************** //
@@ -419,6 +419,18 @@ class StreamTest extends TestCase
        fclose($handle);
    }

+    /**
+     * @return void
+     */
+    public function testWriteEmptyData(): void
+    {
+        $handle = fopen('php://temp', 'rb+');
+        $stream = new Stream($handle);
+
+        $this->expectException(RuntimeException::class);
+        $stream->write('');
+    }
+
    public function writableModeProvider(): array
    {
        return [
@@ -129,19 +129,16 @@ class StreamConnectionTest extends PredisConnectionTestCase
            ->withAnyParameters()
            ->willReturn($this->mockStream);

+        // All handshake commands should be pipelined in a single write
+        $pipelinedCommands = $command1->serializeCommand()
+            . $command2->serializeCommand()
+            . $command3->serializeCommand();
+
        $this->mockStream
-            ->expects($this->exactly(3))
+            ->expects($this->once())
            ->method('write')
-            ->withConsecutive(
-                [$command1->serializeCommand()],
-                [$command2->serializeCommand()],
-                [$command3->serializeCommand()]
-            )
-            ->willReturnOnConsecutiveCalls(
-                strlen($command1->serializeCommand()),
-                strlen($command2->serializeCommand()),
-                strlen($command3->serializeCommand())
-            );
+            ->with($pipelinedCommands)
+            ->willReturn(strlen($pipelinedCommands));

        $this->mockStream
            ->expects($this->exactly(3))
@@ -158,6 +155,61 @@ class StreamConnectionTest extends PredisConnectionTestCase
        $connection->connect();
    }

+    /**
+     * @group disconnected
+     */
+    public function testHandshakeCommandsArePipelinedInSingleNetworkRoundTrip(): void
+    {
+        $parameters = new Parameters();
+        $command1 = new RawCommand('AUTH', ['username', 'password']);
+        $command2 = new RawCommand('SELECT', [5]);
+        $command3 = new RawCommand('CLIENT', ['SETNAME', 'predis']);
+        $command4 = new RawCommand('CLIENT', ['SETINFO', 'LIB-NAME', 'predis']);
+
+        $this->mockStreamFactory
+            ->expects($this->once())
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($this->mockStream);
+
+        // Verify that all handshake commands are serialized and sent together
+        // in a single write operation to reduce initial handshake latency
+        $pipelinedCommands = $command1->serializeCommand()
+            . $command2->serializeCommand()
+            . $command3->serializeCommand()
+            . $command4->serializeCommand();
+
+        $this->mockStream
+            ->expects($this->once())
+            ->method('write')
+            ->with($pipelinedCommands)
+            ->willReturn(strlen($pipelinedCommands));
+
+        // Verify that responses are read separately for each command
+        $this->mockStream
+            ->expects($this->exactly(4))
+            ->method('read')
+            ->with(-1)
+            ->willReturnOnConsecutiveCalls(
+                '+OK\r\n',
+                '+OK\r\n',
+                '+OK\r\n',
+                '+OK\r\n'
+            );
+
+        $connection = new StreamConnection($parameters, $this->mockStreamFactory);
+
+        $connection->addConnectCommand($command1);
+        $connection->addConnectCommand($command2);
+        $connection->addConnectCommand($command3);
+        $connection->addConnectCommand($command4);
+
+        $connection->connect();
+
+        // Verify connection is established
+        $this->assertTrue($connection->isConnected());
+    }
+
    /**
     * @group disconnected
     */
@@ -12,11 +12,15 @@

 namespace Predis\Pipeline;

+use Exception;
 use Predis\Client;
 use Predis\ClientInterface;
 use Predis\Command\Redis\PING;
 use Predis\Connection\Parameters;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
+use Predis\TimeoutException;
 use PredisTestCase;

 class AtomicTest extends PredisTestCase
@@ -56,7 +60,7 @@ class AtomicTest extends PredisTestCase
            );

        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(4))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -104,6 +108,10 @@ class AtomicTest extends PredisTestCase
                $queued,
                $queued
            );
+        $connection
+            ->expects($this->exactly(3))
+            ->method('getParameters')
+            ->willReturn(new Parameters(['protocol' => 2]));

        $pipeline = new Atomic(new Client($connection));

@@ -150,6 +158,10 @@ class AtomicTest extends PredisTestCase
                $queued,
                $error
            );
+        $connection
+            ->expects($this->exactly(3))
+            ->method('getParameters')
+            ->willReturn(new Parameters(['protocol' => 2]));

        $pipeline = new Atomic(new Client($connection));

@@ -191,6 +203,10 @@ class AtomicTest extends PredisTestCase
            ->willReturn(
                new Response\Error('ERR Test error')
            );
+        $connection
+            ->expects($this->exactly(3))
+            ->method('getParameters')
+            ->willReturn(new Parameters(['protocol' => 2]));

        $pipeline = new Atomic(new Client($connection));

@@ -236,7 +252,7 @@ class AtomicTest extends PredisTestCase
            );

        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(4))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -265,6 +281,71 @@ class AtomicTest extends PredisTestCase
        $pipeline->execute();
    }

+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryStandalonePipelineOnRetryableErrors(): void
+    {
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+        $mockConnection = $this->getMockConnection();
+
+        $mockConnection
+            ->expects($this->exactly(2))
+            ->method('executeCommand')
+            ->withConsecutive(
+                [$this->isRedisCommand('MULTI')],
+                [$this->isRedisCommand('EXEC')]
+            )
+            ->willReturnOnConsecutiveCalls(
+                new Response\Status('OK'),
+                ['PONG', 'PONG', 'PONG']
+            );
+
+        $mockConnection
+            ->expects($this->exactly(4))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                1000
+            );
+
+        $mockConnection
+            ->expects($this->exactly(3))
+            ->method('readResponse')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                "+QUEUED\r\n",
+                "+QUEUED\r\n",
+                "+QUEUED\r\n"
+            );
+
+        $mockConnection
+            ->expects($this->atLeast(3))
+            ->method('disconnect')
+            ->withAnyParameters();
+
+        $mockConnection
+            ->expects($this->exactly(4))
+            ->method('getParameters')
+            ->willReturn($parameters);
+
+        $pipeline = new Atomic(new Client($mockConnection));
+
+        $responses = $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->ping();
+            $pipe->ping();
+            $pipe->ping();
+        });
+
+        $this->assertEquals(['PONG', 'PONG', 'PONG'], $responses);
+    }
+
    /**
     * @group connected
     * @group relay-incompatible
@@ -273,7 +354,7 @@ class AtomicTest extends PredisTestCase
    {
        $parameters = $this->getDefaultParametersArray();

-        $client = $this->getClient(
+        $client = new Client(
            ["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
            ['replication' => 'predis']
        );
@@ -12,10 +12,17 @@

 namespace Predis\Pipeline;

+use Exception;
 use Predis\Client;
 use Predis\ClientInterface;
 use Predis\Command\Redis\PING;
+use Predis\Connection\Cluster\RedisCluster;
+use Predis\Connection\Parameters;
+use Predis\Connection\Replication\MasterSlaveReplication;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
+use Predis\TimeoutException;
 use PredisTestCase;

 class FireAndForgetTest extends PredisTestCase
@@ -34,6 +41,10 @@ class FireAndForgetTest extends PredisTestCase
        $connection
            ->expects($this->never())
            ->method('readResponse');
+        $connection
+            ->expects($this->exactly(1))
+            ->method('getParameters')
+            ->willReturn(new Parameters(['protocol' => 2]));

        $pipeline = new FireAndForget(new Client($connection));

@@ -68,6 +79,11 @@ class FireAndForgetTest extends PredisTestCase
            ->expects($this->never())
            ->method('readResponse');

+        $connection
+            ->expects($this->exactly(2))
+            ->method('getParameters')
+            ->willReturn(new Parameters(['protocol' => 2]));
+
        $pipeline = new FireAndForget(new Client($connection));

        $pipeline->ping();
@@ -77,6 +93,147 @@ class FireAndForgetTest extends PredisTestCase
        $this->assertEmpty($pipeline->execute());
    }

+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryStandalonePipelineOnRetryableErrors(): void
+    {
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+        $mockConnection = $this->getMockConnection();
+
+        $mockConnection
+            ->expects($this->exactly(4))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                1000
+            );
+
+        $mockConnection
+            ->expects($this->atLeast(3))
+            ->method('disconnect')
+            ->withAnyParameters();
+
+        $mockConnection
+            ->expects($this->exactly(1))
+            ->method('getParameters')
+            ->willReturn($parameters);
+
+        $pipeline = new FireAndForget(new Client($mockConnection));
+
+        $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->ping();
+            $pipe->ping();
+            $pipe->ping();
+        });
+    }
+
+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryClusterPipelineOnRetryableErrors(): void
+    {
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+        $mockConnection = $this->getMockConnection();
+        $mockClusterConnection = $this->getMockBuilder(RedisCluster::class)
+            ->disableOriginalConstructor()->getMock();
+
+        $mockConnection
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                1000,
+                1000,
+                1000
+            );
+
+        $mockConnection
+            ->expects($this->atLeast(3))
+            ->method('disconnect')
+            ->withAnyParameters();
+
+        $mockClusterConnection
+            ->expects($this->exactly(5))
+            ->method('getParameters')
+            ->willReturn($parameters);
+
+        $mockClusterConnection
+            ->expects($this->exactly(6))
+            ->method('getConnectionByCommand')
+            ->willReturn($mockConnection);
+
+        $pipeline = new FireAndForget(new Client($mockClusterConnection));
+
+        $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->ping();
+            $pipe->ping();
+            $pipe->ping();
+        });
+    }
+
+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryReplicationPipelineOnRetryableErrors(): void
+    {
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+        $mockConnection = $this->getMockConnection();
+        $mockReplicationConnection = $this->getMockBuilder(MasterSlaveReplication::class)
+            ->disableOriginalConstructor()->getMock();
+
+        $mockConnection
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                1000,
+                1000,
+                1000
+            );
+
+        $mockConnection
+            ->expects($this->atLeast(3))
+            ->method('disconnect')
+            ->withAnyParameters();
+
+        $mockReplicationConnection
+            ->expects($this->exactly(5))
+            ->method('getParameters')
+            ->willReturn($parameters);
+
+        $mockReplicationConnection
+            ->expects($this->exactly(6))
+            ->method('getConnectionByCommand')
+            ->willReturn($mockConnection);
+
+        $pipeline = new FireAndForget(new Client($mockReplicationConnection));
+
+        $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->ping();
+            $pipe->ping();
+            $pipe->ping();
+        });
+    }
+
    /**
     * @group connected
     * @group cluster
@@ -105,7 +262,7 @@ class FireAndForgetTest extends PredisTestCase
    {
        $parameters = $this->getDefaultParametersArray();

-        $client = $this->getClient(
+        $client = new Client(
            ["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
            ['replication' => 'predis']
        );
@@ -20,9 +20,20 @@ use Predis\ClientInterface;
 use Predis\Command\CommandInterface;
 use Predis\Command\Redis\ECHO_;
 use Predis\Command\Redis\PING;
+use Predis\Connection\Cluster\RedisCluster;
+use Predis\Connection\Factory;
 use Predis\Connection\Parameters;
+use Predis\Connection\Replication\MasterSlaveReplication;
+use Predis\Connection\Resource\StreamFactoryInterface;
+use Predis\Connection\StreamConnection;
+use Predis\Replication\ReplicationStrategy;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
+use Predis\TimeoutException;
 use PredisTestCase;
+use Psr\Http\Message\StreamInterface;
+use RuntimeException;
 use stdClass;

 class PipelineTest extends PredisTestCase
@@ -91,7 +102,7 @@ class PipelineTest extends PredisTestCase
            ->willReturn($object);

        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(2))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -119,7 +130,7 @@ class PipelineTest extends PredisTestCase
            ->willReturn($error);

        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(2))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -145,7 +156,7 @@ class PipelineTest extends PredisTestCase
            ->willReturn($error);

        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(2))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -238,7 +249,7 @@ class PipelineTest extends PredisTestCase
            ->willReturnCallback($this->getReadCallback());

        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(2))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -301,7 +312,7 @@ class PipelineTest extends PredisTestCase
            ->willReturnCallback($this->getReadCallback());

        $connection
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(4))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -333,15 +344,15 @@ class PipelineTest extends PredisTestCase
            ->expects($this->once())
            ->method('switchToMaster');
        $connection
-            ->expects($this->exactly(3))
+            ->expects($this->exactly(6))
            ->method('getConnectionByCommand')
            ->willReturn($nodeConnection);
-        $connection
+        $nodeConnection
            ->expects($this->exactly(3))
            ->method('readResponse')
            ->willReturn($pong);
        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(3))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -429,7 +440,7 @@ class PipelineTest extends PredisTestCase
            ->method('readResponse')
            ->willReturnCallback($this->getReadCallback());
        $connection
-            ->expects($this->once())
+            ->expects($this->exactly(2))
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

@@ -478,6 +489,327 @@ class PipelineTest extends PredisTestCase
        $this->assertNull($responses);
    }

+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryStandalonePipelineOnRetryableErrors(): void
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $mockStream
+            ->expects($this->atLeast(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(4))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000
+            );
+
+        $mockStream
+            ->expects($this->exactly(3))
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+PONG\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection = new StreamConnection($parameters, $mockStreamFactory);
+        $pipeline = new Pipeline(new Client($connection));
+
+        $responses = $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->ping();
+            $pipe->ping();
+            $pipe->ping();
+        });
+
+        $this->assertEquals(['PONG', 'PONG', 'PONG'], $responses);
+    }
+
+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryClusterPipelineOnRetryableErrors(): void
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $mockConnectionFactory = $this->getMockBuilder(Factory::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $mockStream
+            ->expects($this->atLeast(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000,
+                1000,
+                1000
+            );
+
+        $mockStream
+            ->expects($this->exactly(3))
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $streamConnection = new StreamConnection($parameters, $mockStreamFactory);
+        $connection = new RedisCluster($mockConnectionFactory, $parameters);
+        $connection->add($streamConnection);
+
+        $pipeline = new Pipeline(new Client($connection));
+
+        $responses = $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+        });
+
+        $this->assertEquals(['OK', 'OK', 'OK'], $responses);
+    }
+
+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testExecutePipelineInvokesOnAggregateConnectionFailCallbackOnConnectionException(): void
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $mockConnectionFactory = $this->getMockBuilder(Factory::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $streamConnection = new StreamConnection($parameters, $mockStreamFactory);
+        $connection = $this->getMockBuilder(RedisCluster::class)
+            ->setConstructorArgs([$mockConnectionFactory, $parameters])
+            ->onlyMethods(['getConnectionByCommand', 'remove', 'askSlotMap'])
+            ->getMock();
+
+        // Disable useClusterSlots to avoid askSlotMap() calls
+        $connection->useClusterSlots = false;
+
+        $mockStream
+            ->expects($this->atLeast(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new \Predis\Connection\ConnectionException($streamConnection, 'Connection failed')),
+                $this->throwException(new \Predis\Connection\ConnectionException($streamConnection, 'Connection failed')),
+                $this->throwException(new \Predis\Connection\ConnectionException($streamConnection, 'Connection failed')),
+                1000,
+                1000,
+                1000
+            );
+
+        $mockStream
+            ->expects($this->exactly(3))
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection
+            ->expects($this->exactly(9))
+            ->method('getConnectionByCommand')
+            ->willReturn($streamConnection);
+
+        // Verify that remove() is called on the aggregate connection during retry
+        $connection
+            ->expects($this->exactly(3))
+            ->method('remove')
+            ->with($streamConnection);
+
+        // Verify that askSlotMap() is NOT called since useClusterSlots is false
+        $connection
+            ->expects($this->never())
+            ->method('askSlotMap');
+
+        $pipeline = new Pipeline(new Client($connection));
+
+        $responses = $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+        });
+
+        $this->assertEquals(['OK', 'OK', 'OK'], $responses);
+    }
+
+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testExecutePipelineInvokesOnAggregateConnectionFailCallbackOnTimeoutException(): void
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $mockConnectionFactory = $this->getMockBuilder(Factory::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $streamConnection = new StreamConnection($parameters, $mockStreamFactory);
+        $connection = $this->getMockBuilder(RedisCluster::class)
+            ->setConstructorArgs([$mockConnectionFactory, $parameters])
+            ->onlyMethods(['getConnectionByCommand', 'remove'])
+            ->getMock();
+
+        // Disable useClusterSlots to avoid askSlotMap() calls
+        $connection->useClusterSlots = false;
+
+        $mockStream
+            ->expects($this->atLeast(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($streamConnection, 0)),
+                $this->throwException(new TimeoutException($streamConnection, 0)),
+                $this->throwException(new TimeoutException($streamConnection, 0)),
+                1000,
+                1000,
+                1000
+            );
+
+        $mockStream
+            ->expects($this->exactly(3))
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $connection
+            ->expects($this->exactly(9))
+            ->method('getConnectionByCommand')
+            ->willReturn($streamConnection);
+
+        // Verify that remove() is NOT called for TimeoutException
+        $connection
+            ->expects($this->never())
+            ->method('remove');
+
+        $pipeline = new Pipeline(new Client($connection));
+
+        $responses = $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+        });
+
+        $this->assertEquals(['OK', 'OK', 'OK'], $responses);
+    }
+
+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryReplicationPipelineOnRetryableErrors(): void
+    {
+        $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
+        $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+            'role' => 'master',
+        ]);
+
+        $mockStream
+            ->expects($this->atLeast(3))
+            ->method('close')
+            ->withAnyParameters();
+
+        $mockStream
+            ->expects($this->exactly(6))
+            ->method('write')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                $this->throwException(new RuntimeException('', 2)),
+                1000,
+                1000,
+                1000
+            );
+
+        $mockStream
+            ->expects($this->exactly(3))
+            ->method('read')
+            ->withAnyParameters()
+            ->willReturn("+OK\r\n");
+
+        $mockStreamFactory
+            ->expects($this->exactly(4))
+            ->method('createStream')
+            ->withAnyParameters()
+            ->willReturn($mockStream);
+
+        $streamConnection = new StreamConnection($parameters, $mockStreamFactory);
+
+        $connection = new MasterSlaveReplication(new ReplicationStrategy());
+        $connection->add($streamConnection);
+
+        $pipeline = new Pipeline(new Client($connection));
+
+        $responses = $pipeline->execute(function (Pipeline $pipe) {
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+            $pipe->set('key', 'value');
+        });
+
+        $this->assertEquals(['OK', 'OK', 'OK'], $responses);
+    }
+
    // ******************************************************************** //
    // ---- INTEGRATION TESTS --------------------------------------------- //
    // ******************************************************************** //
@@ -648,6 +980,54 @@ class PipelineTest extends PredisTestCase
        $this->assertSameValues($expectedResults, $results);
    }

+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 6.2.0
+     * @return void
+     */
+    public function testStandaloneRetryPipelineOnTimeoutException(): void
+    {
+        $client = $this->getClient([
+            'retry' => new Retry(new ExponentialBackoff(100, 1000), 3),
+            'read_write_timeout' => 0.1,
+        ]);
+
+        $this->expectException(TimeoutException::class);
+
+        $client->pipeline(function (Pipeline $pipe) use (&$retries) {
+            $pipe->incr('test_key');
+            $pipe->blpop('foo', 3);
+        });
+        $this->assertEquals(3, $client->get('test_key'));
+    }
+
+    /**
+     * @group connected
+     * @group cluster
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 6.2.0
+     * @return void
+     */
+    public function testClusterRetryPipelineOnTimeoutException(): void
+    {
+        $retries = 0;
+        $client = $this->getClient([], [
+            'parameters' => [
+                'retry' => new Retry(new ExponentialBackoff(100, 1000), 3),
+                'read_write_timeout' => 0.1,
+            ],
+        ]);
+
+        $this->expectException(TimeoutException::class);
+
+        $client->pipeline(function (Pipeline $pipe) use (&$retries) {
+            ++$retries;
+            $pipe->blpop('foo', 3);
+        });
+        $this->assertEquals(3, $retries);
+    }
+
    /**
     * @group connected
     * @group relay-incompatible
@@ -656,7 +1036,7 @@ class PipelineTest extends PredisTestCase
    {
        $parameters = $this->getDefaultParametersArray();

-        $client = $this->getClient(
+        $client = new Client(
            ["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
            ['replication' => 'predis']
        );
@@ -0,0 +1,154 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry;
+
+use PHPUnit\Framework\TestCase;
+use Predis\Connection\ConnectionException;
+use Predis\Connection\NodeConnectionInterface;
+use Predis\Connection\Resource\Exception\StreamInitException;
+use Predis\Retry\Strategy\EqualBackoff;
+use Predis\Retry\Strategy\ExponentialBackoff;
+use Predis\Retry\Strategy\NoBackoff;
+use Predis\Retry\Strategy\RetryStrategyInterface;
+use RuntimeException;
+use Throwable;
+
+class RetryTest extends TestCase
+{
+    /**
+     * @group disconnected
+     * @dataProvider strategyProvider
+     * @throws Throwable
+     */
+    public function testCallWithRetry(
+        RetryStrategyInterface $backoffStrategy,
+        int $retries,
+        float $expectedExecutionTime,
+        float $delta
+    ) {
+        $retry = new Retry($backoffStrategy, $retries);
+        $retriesCount = 0;
+
+        $callable = function () use (&$retriesCount, $retries) {
+            if ($retriesCount >= $retries) {
+                return;
+            }
+
+            ++$retriesCount;
+            throw new StreamInitException();
+        };
+
+        $startTime = microtime(true);
+        $retry->callWithRetry($callable);
+        $executionTime = microtime(true) - $startTime;
+
+        $this->assertEquals($retriesCount, $retries);
+        $this->assertEqualsWithDelta($expectedExecutionTime, $executionTime, $delta);
+
+        $retry->updateRetriesCount(10);
+        $this->assertEquals(10, $retry->getRetries());
+    }
+
+    /**
+     * @group disconnected
+     * @return void
+     */
+    public function testNoRetriesOnExcludedRetryableExceptions()
+    {
+        $retry = new Retry(new NoBackoff(), 3, [ConnectionException::class]);
+        $retriesCount = 0;
+        $callCount = 0;
+
+        $doCallable = function () use (&$callCount) {
+            ++$callCount;
+
+            if ($callCount <= 3) {
+                throw new RuntimeException();
+            } elseif ($callCount <= 7) {
+                throw new ConnectionException(
+                    $this->getMockBuilder(NodeConnectionInterface::class)->getMock()
+                );
+            } else {
+                throw new StreamInitException();
+            }
+        };
+
+        $failCallable = function () use (&$retriesCount) {
+            ++$retriesCount;
+        };
+
+        // Ensures that no retries happens on excluded exception.
+        while ($callCount < 3) {
+            try {
+                $retry->callWithRetry($doCallable, $failCallable);
+            } catch (Throwable $e) {
+                $this->assertInstanceOf(RuntimeException::class, $e);
+                $this->assertEquals(0, $retriesCount);
+            }
+        }
+
+        // Ensures that retries happens on specified exception.
+        try {
+            $retry->callWithRetry($doCallable, $failCallable);
+        } catch (Throwable $e) {
+            $this->assertInstanceOf(ConnectionException::class, $e);
+            $this->assertEquals(3, $retriesCount);
+        }
+
+        $retry->updateCatchableExceptions([StreamInitException::class]);
+
+        // Ensures that retries happens on updated catchable exceptions.
+        try {
+            $retry->callWithRetry($doCallable, $failCallable);
+        } catch (Throwable $e) {
+            $this->assertInstanceOf(StreamInitException::class, $e);
+            $this->assertEquals(6, $retriesCount);
+        }
+
+        $this->assertEquals(11, $callCount);
+    }
+
+    public function strategyProvider(): array
+    {
+        return [
+            'NoBackoff' => [
+                new NoBackoff(),
+                3,
+                1,
+                1,
+            ],
+            'EqualBackoff' => [
+                new EqualBackoff(0.3 * 1000000),
+                3,
+                0.9,
+                0.1,
+            ],
+            'ExponentialBackoff - no jitter' => [
+                new ExponentialBackoff(),
+                3,
+                0.112,
+                0.08,
+            ],
+            'ExponentialBackoff - with jitter' => [
+                new ExponentialBackoff(
+                    RetryStrategyInterface::DEFAULT_BASE,
+                    RetryStrategyInterface::DEFAULT_CAP,
+                    true
+                ),
+                3,
+                0.112,
+                0.112, // Theoretically, jitter==0 might happen sequentially 3 times
+            ],
+        ];
+    }
+}
@@ -0,0 +1,28 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+use PHPUnit\Framework\TestCase;
+
+class EqualBackoffTest extends TestCase
+{
+    /**
+     * @group disconnected
+     * @return void
+     */
+    public function testCompute(): void
+    {
+        $backoff = new EqualBackoff(1);
+        $this->assertEquals(1, $backoff->compute(1));
+    }
+}
@@ -0,0 +1,70 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+use PHPUnit\Framework\TestCase;
+
+class ExponentialBackoffTest extends TestCase
+{
+    /**
+     * @group disconnected
+     * @return void
+     */
+    public function testCompute(): void
+    {
+        $backoff = new ExponentialBackoff();
+
+        // Test default cap
+        $this->assertLessThanOrEqual(RetryStrategyInterface::DEFAULT_CAP, $backoff->compute(100));
+
+        // Test default base
+        $this->assertGreaterThanOrEqual(RetryStrategyInterface::DEFAULT_BASE, $backoff->compute(0));
+
+        $interval = $backoff->compute(2);
+
+        // Test between
+        $this->assertGreaterThanOrEqual(RetryStrategyInterface::DEFAULT_BASE, $interval);
+        $this->assertLessThanOrEqual(RetryStrategyInterface::DEFAULT_CAP, $interval);
+
+        $backoff = new ExponentialBackoff(1000000, 10000000);
+
+        // Test adjusted cap
+        $this->assertLessThanOrEqual(10000000, $backoff->compute(100));
+
+        // Test adjusted base
+        $this->assertGreaterThanOrEqual(1000000, $backoff->compute(0));
+
+        $backoff = new ExponentialBackoff(RetryStrategyInterface::DEFAULT_BASE, -1);
+
+        // Test with no cap
+        $this->assertEquals(RetryStrategyInterface::DEFAULT_BASE * 2, $backoff->compute(1));
+
+        $backoff = new ExponentialBackoff(
+            RetryStrategyInterface::DEFAULT_BASE,
+            RetryStrategyInterface::DEFAULT_CAP,
+            true
+        );
+
+        $interval = $backoff->compute(0);
+
+        // Test with jitter - default base
+        $this->assertGreaterThanOrEqual(0, $interval);
+        $this->assertLessThanOrEqual(RetryStrategyInterface::DEFAULT_BASE, $interval);
+
+        $interval = $backoff->compute(6);
+
+        // Test with jitter - default cap
+        $this->assertGreaterThanOrEqual(0, $interval);
+        $this->assertLessThanOrEqual(RetryStrategyInterface::DEFAULT_CAP, $interval);
+    }
+}
@@ -0,0 +1,28 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Retry\Strategy;
+
+use PHPUnit\Framework\TestCase;
+
+class NoBackoffTest extends TestCase
+{
+    /**
+     * @group disconnected
+     * @return void
+     */
+    public function testCompute(): void
+    {
+        $backoff = new NoBackoff();
+        $this->assertEquals(0, $backoff->compute(1));
+    }
+}
@@ -0,0 +1,126 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2025 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis;
+
+use Predis\Connection\Resource\Exception\StreamInitException;
+use PredisTestCase;
+
+class SSLTest extends PredisTestCase
+{
+    /**
+     * @group connected
+     * @group ssl
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.2.0
+     * @return void
+     */
+    public function testExecuteCommandOverSSLConnection()
+    {
+        $redis = $this->createClient();
+        $this->assertEquals('PONG', $redis->ping());
+    }
+
+    /**
+     * @group connected
+     * @group ssl
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.2.0
+     * @return void
+     */
+    public function testExecuteCommandOverSSLConnectionFailsOnIncorrectCertificate()
+    {
+        $redis = new Client($this->getDefaultParametersArray() + [
+            'ssl' => ['cafile' => '/tmp/invalid.crt', 'verify_peer' => true, 'verify_peer_name' => false]]
+        );
+
+        $this->expectException(StreamInitException::class);
+        $this->expectExceptionMessage('Error while switching to encrypted communication');
+
+        $redis->ping();
+    }
+
+    /**
+     * @group connected
+     * @group ssl
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.2.0
+     * @return void
+     */
+    public function testExecuteCommandOverSSLConnectionWithoutSSLConfig()
+    {
+        $redis = new Client($this->getDefaultParametersArray());
+
+        $this->expectException(StreamInitException::class);
+        $this->expectExceptionMessage('Error while switching to encrypted communication');
+
+        $redis->ping();
+    }
+
+    /**
+     * @group connected
+     * @group ssl
+     * @group cluster
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.2.0
+     * @return void
+     */
+    public function testClusterExecuteCommandOverSSLConnection()
+    {
+        $redis = $this->createClient();
+        $redis->set('foo', 'bar');
+        $this->assertEquals('bar', $redis->get('foo'));
+    }
+
+    /**
+     * @group connected
+     * @group ssl
+     * @group cluster
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.2.0
+     * @return void
+     */
+    public function testClusterExecuteCommandOverSSLConnectionFailsOnIncorrectCertificate()
+    {
+        $redis = new Client($this->getDefaultParametersArray(), [
+            'cluster' => 'redis',
+            'parameters' => [
+                'ssl' => ['cafile' => '/tmp/invalid.crt', 'verify_peer' => true, 'verify_peer_name' => false],
+            ],
+        ]);
+
+        $this->expectException(StreamInitException::class);
+        $this->expectExceptionMessage('Error while switching to encrypted communication');
+
+        $redis->set('foo', 'bar');
+    }
+
+    /**
+     * @group connected
+     * @group ssl
+     * @group cluster
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.2.0
+     * @return void
+     */
+    public function testClusterExecuteCommandOverSSLConnectionWithoutSSLConfig()
+    {
+        $redis = new Client($this->getDefaultParametersArray(), [
+            'cluster' => 'redis',
+        ]);
+
+        $this->expectException(StreamInitException::class);
+        $this->expectExceptionMessage('Error while switching to encrypted communication');
+
+        $redis->set('foo', 'bar');
+    }
+}
@@ -20,6 +20,9 @@ use Predis\Command\CommandInterface;
 use Predis\Connection\NodeConnectionInterface;
 use Predis\Connection\Parameters;
 use Predis\Response;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
+use Predis\TimeoutException;
 use Predis\Transaction\Exception\TransactionException;
 use PredisTestCase;
 use RuntimeException;
@@ -675,6 +678,55 @@ class MultiExecTest extends PredisTestCase
        $tx->multi()->echo('test')->exec();
    }

+    /**
+     * @group disconnected
+     * @throws Exception
+     */
+    public function testRetryReplicationPipelineOnRetryableErrors(): void
+    {
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+            'role' => 'master',
+        ]);
+
+        $mockConnection = $this->getMockConnection();
+
+        $mockConnection
+            ->expects($this->any())
+            ->method('getParameters')
+            ->willReturn($parameters);
+
+        $mockConnection
+            ->expects($this->atLeast(3))
+            ->method('disconnect')
+            ->withAnyParameters();
+
+        $mockConnection
+            ->expects($this->exactly(8))
+            ->method('executeCommand')
+            ->withAnyParameters()
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                $this->throwException(new TimeoutException($mockConnection)),
+                new Response\Status('OK'),
+                new Response\Status('QUEUED'),
+                new Response\Status('QUEUED'),
+                new Response\Status('QUEUED'),
+                ['OK', 'OK', 'OK']
+            );
+
+        $tx = new MultiExec(new Client($mockConnection));
+
+        $responses = $tx->execute(function (MultiExec $tx) {
+            $tx->set('key', 'value');
+            $tx->set('key', 'value');
+            $tx->set('key', 'value');
+        });
+
+        $this->assertEquals(['OK', 'OK', 'OK'], $responses);
+    }
+
    // ******************************************************************** //
    // ---- INTEGRATION TESTS --------------------------------------------- //
    // ******************************************************************** //
@@ -15,6 +15,10 @@ namespace Predis\Transaction\Strategy;
 use PHPUnit\Framework\TestCase;
 use Predis\Command\CommandInterface;
 use Predis\Connection\NodeConnectionInterface;
+use Predis\Connection\Parameters;
+use Predis\Retry\Retry;
+use Predis\Retry\Strategy\ExponentialBackoff;
+use Predis\TimeoutException;
 use Predis\Transaction\MultiExecState;

 class NodeConnectionStrategyTest extends TestCase
@@ -46,8 +50,72 @@ class NodeConnectionStrategyTest extends TestCase
            ->with($this->mockCommand)
            ->willReturn('OK');

+        $this->mockConnection
+            ->expects($this->any())
+            ->method('getParameters')
+            ->willReturn(new Parameters());
+
        $strategy = new NodeConnectionStrategy($this->mockConnection, new MultiExecState());

        $this->assertEquals('OK', $strategy->executeCommand($this->mockCommand));
    }
+
+    /**
+     * @return void
+     */
+    public function testUnwatch(): void
+    {
+        $this->mockConnection
+            ->expects($this->once())
+            ->method('executeCommand')
+            ->with($this->callback(function ($command) {
+                return $command->getId() === 'UNWATCH';
+            }))
+            ->willReturn('OK');
+
+        $this->mockConnection
+            ->expects($this->any())
+            ->method('getParameters')
+            ->willReturn(new Parameters());
+
+        $strategy = new NodeConnectionStrategy($this->mockConnection, new MultiExecState());
+
+        $this->assertEquals('OK', $strategy->unwatch());
+    }
+
+    /**
+     * @return void
+     */
+    public function testUnwatchWithRetries(): void
+    {
+        $parameters = new Parameters([
+            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
+        ]);
+
+        $this->mockConnection
+            ->expects($this->exactly(4))
+            ->method('executeCommand')
+            ->with($this->callback(function ($command) {
+                return $command->getId() === 'UNWATCH';
+            }))
+            ->willReturnOnConsecutiveCalls(
+                $this->throwException(new TimeoutException($this->mockConnection)),
+                $this->throwException(new TimeoutException($this->mockConnection)),
+                $this->throwException(new TimeoutException($this->mockConnection)),
+                'OK'
+            );
+
+        $this->mockConnection
+            ->expects($this->any())
+            ->method('getParameters')
+            ->willReturn($parameters);
+
+        $this->mockConnection
+            ->expects($this->exactly(3))
+            ->method('disconnect');
+
+        $strategy = new NodeConnectionStrategy($this->mockConnection, new MultiExecState());
+
+        $this->assertEquals('OK', $strategy->unwatch());
+    }
 }
@@ -14,6 +14,7 @@ namespace Predis\Transaction\Strategy;

 use PHPUnit\Framework\TestCase;
 use Predis\Command\CommandInterface;
+use Predis\Connection\Parameters;
 use Predis\Connection\Replication\ReplicationInterface;
 use Predis\Transaction\MultiExecState;

@@ -46,6 +47,11 @@ class ReplicationConnectionStrategyTest extends TestCase
            ->with($this->mockCommand)
            ->willReturn('OK');

+        $this->mockConnection
+            ->expects($this->any())
+            ->method('getParameters')
+            ->willReturn(new Parameters());
+
        $strategy = new ReplicationConnectionStrategy($this->mockConnection, new MultiExecState());

        $this->assertEquals('OK', $strategy->executeCommand($this->mockCommand));
Author	SHA1	Message	Date
vladvildanov	ea9e26e0ba	Testing with pre 8.6 build	2026-01-21 10:48:00 +02:00
Vladyslav Vildanov	95e802430d	Added testing with SSL connection (#1624 ) * Added testing with SSL connection * Update CHANGELOG.md * Revert changes * Codestyle fixes * Add version restriction * Fixed issue with connecting to SSL port to check redis version * Mark tests as relay-incompatible * Codestyle fixes	2026-01-20 17:45:10 +02:00
Vladyslav Vildanov	0703c3cae6	Added support for VRANGE command (#1623 ) * Added support for VRANGE command * Update CHANGELOG.md * Codestyle fixes * Codestyle fixes	2026-01-19 10:50:39 +02:00
Vladyslav Vildanov	7c7cf90152	Improve connection handshake by pipelining commands (#1622 ) * Improve connection handshake by pipelining commands * Added exceptions handling for Redis < 6.0 * Updated CHANGELOG.md	2026-01-08 13:56:55 +02:00
Markus Reinhold	b211afd755	Make ZRANDMEMBER prefixable (#1621 ) * Make ZRANDMEMBER prefixable * Update CHANGELOG.md --------- Co-authored-by: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com>	2025-12-23 13:27:27 +02:00
Vladyslav Vildanov	980326d5d8	Added retry support (#1616 ) * Initial work on retries * Added retry class and test coverage * Added support for standalone and cluster * Make TimeoutException instance of CommunicationException * Added pipeline, trasnaction, replication support * Fixed broken test * Marked test as relay-incompatible * Marked test as relay-incompatible * Fixed analysis errors, added missing tests * Codestyle fixes * Fixed test * Update README.md * Update README.md * Update README.md * Updated README.md * Refactor retry on read and write * Added check for timeout value * Updated README.md * Fixed README.md * Codestyle changes * Added missing coverage * Added missing test coverage * Removed comments * Added retry support for Relay connection (#1620) * Added integration test case with mocked retry * Changed client initialisation in tests * Marked test as relay-incompatible --------- Co-authored-by: Pavlo Yatsukhnenko <yatsukhnenko@users.noreply.github.com>	2025-12-23 10:40:46 +02:00
Pavlo Yatsukhnenko	0ae180c942	Enable more Relay tests (#1617 ) run tests against relay nightly for faster testing	2025-11-27 16:18:22 -08:00
Andrew Ivchenkov	98c853a689	fixed param annotation (#1614 )	2025-11-25 09:04:34 -08:00