20455852140

Committed 23 Dec 2025 08:40AM UTC coverage: 92.615% (-0.2%) from 92.788%

Build # 20455852140

Build Type

push

github

Committed by

web-flow

Commit Message

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>

Run Details

301 of 358 new or added lines in 18 files covered. (84.08%)

2 existing lines in 2 files now uncovered.

8214 of 8869 relevant lines covered (92.61%)

112.31 hits per line

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

73.91

/src/Transaction/Strategy/NonClusterConnectionStrategy.php

<?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\Transaction\Strategy;

use Predis\Command\CommandInterface;
use Predis\Command\Redis\DISCARD;
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.
 */
abstract class NonClusterConnectionStrategy implements StrategyInterface
{
    /**
     * @var NodeConnectionInterface|ReplicationInterface
     */
    protected $connection;

    /**
     * @var MultiExecState
     */
    protected $state;

    /**
     * @param NodeConnectionInterface|ReplicationInterface $connection
     */
    public function __construct($connection, MultiExecState $state)
    {
        $this->connection = $connection;
        $this->state = $state;
    }

    /**
     * {@inheritDoc}
     */
    public function initializeTransaction(): bool
    {
        return 'OK' == $this->executeBypassingTransaction(new MULTI())->getResponse();
    }

    /**
     * {@inheritDoc}
     * @throws Throwable
     */
    public function executeCommand(CommandInterface $command)
    {
        if ($this->state->isCAS()) {
            return $this->executeBypassingTransaction($command);
        }

        $retry = $this->connection->getParameters()->retry;

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

    /**
     * {@inheritDoc}
     */
    public function executeTransaction()
    {
        return $this->executeBypassingTransaction(new EXEC())->getResponse();
    }

    /**
     * {@inheritDoc}
     */
    public function multi()
    {
        return $this->executeBypassingTransaction(new MULTI())->getResponse();
    }

    /**
     * {@inheritDoc}
     */
    public function watch(array $keys)
    {
        $watch = new WATCH();
        $watch->setArguments($keys);

        return $this->executeBypassingTransaction($watch)->getResponse();
    }

    /**
     * {@inheritDoc}
     * @throws Throwable
     */
    public function unwatch()
    {
        $retry = $this->connection->getParameters()->retry;

        return $retry->callWithRetry(
            function () {
                return $this->connection->executeCommand(new UNWATCH());
            }, function (CommunicationException $e) {
                $this->onFailCallback($e);
            }
        );
    }

    /**
     * {@inheritDoc}
     */
    public function discard()
    {
        return $this->executeBypassingTransaction(new DISCARD())->getResponse();
    }

    /**
     * Executes a Redis command bypassing the transaction logic.
     *
     * @param  CommandInterface          $command
     * @return BypassTransactionResponse
     * @throws ServerException|Throwable
     */
    protected function executeBypassingTransaction(CommandInterface $command): BypassTransactionResponse
    {
        $retry = $this->connection->getParameters()->retry;

        try {
            $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;
            }

            if (strcasecmp($command->getId(), 'EXEC') != 0) {
                throw $exception;
            }

            if (!strpos($exception->getMessage(), 'RELAY_ERR_REDIS')) {
                throw $exception;
            }

            return new BypassTransactionResponse(null);
        }

        if ($response instanceof ErrorInterface) {
            throw new ServerException($response->getMessage());
        }

        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();
            }
        }
    }
}

1	<?php
2
3	/*
4	* This file is part of the Predis package.
5	*
6	* (c) 2009-2020 Daniele Alessandri
7	* (c) 2021-2025 Till Krüss
8	*
9	* For the full copyright and license information, please view the LICENSE
10	* file that was distributed with this source code.
11	*/
12
13	namespace Predis\Transaction\Strategy;
14
15	use Predis\Command\CommandInterface;
16	use Predis\Command\Redis\DISCARD;
17	use Predis\Command\Redis\EXEC;
18	use Predis\Command\Redis\MULTI;
19	use Predis\Command\Redis\UNWATCH;
20	use Predis\Command\Redis\WATCH;
21	use Predis\CommunicationException;
22	use Predis\Connection\ConnectionException;
23	use Predis\Connection\NodeConnectionInterface;
24	use Predis\Connection\RelayConnection;
25	use Predis\Connection\Replication\ReplicationInterface;
26	use Predis\Response\ErrorInterface;
27	use Predis\Response\ServerException;
28	use Predis\TimeoutException;
29	use Predis\Transaction\MultiExecState;
30	use Predis\Transaction\Response\BypassTransactionResponse;
31	use Throwable;
32
33	/**
34	* Defines strategy for connections that operates on non-distributed hash slots.
35	*/
36	abstract class NonClusterConnectionStrategy implements StrategyInterface
37	{
38	/**
39	* @var NodeConnectionInterface\|ReplicationInterface
40	*/
41	protected $connection;
42
43	/**
44	* @var MultiExecState
45	*/
46	protected $state;
47
48	/**
49	* @param NodeConnectionInterface\|ReplicationInterface $connection
50	*/
51	public function __construct($connection, MultiExecState $state)	45✔
52	{
53	$this->connection = $connection;	45✔
54	$this->state = $state;	45✔
55	}
56
57	/**
58	* {@inheritDoc}
59	*/
60	public function initializeTransaction(): bool	26✔
61	{
62	return 'OK' == $this->executeBypassingTransaction(new MULTI())->getResponse();	26✔
63	}
64
65	/**
66	* {@inheritDoc}
67	* @throws Throwable
68	*/
69	public function executeCommand(CommandInterface $command)	32✔
70	{
71	if ($this->state->isCAS()) {	32✔
72	return $this->executeBypassingTransaction($command);	5✔
73	}
74
75	$retry = $this->connection->getParameters()->retry;	30✔
76
77	return $retry->callWithRetry(	30✔
78	function () use ($command) {	30✔
79	return $this->connection->executeCommand($command);	30✔
80	}, function (CommunicationException $e) {	30✔
NEW 81	$this->onFailCallback($e);	×
82	}	30✔
83	);	30✔
84	}
85
86	/**
87	* {@inheritDoc}
88	*/
89	public function executeTransaction()	20✔
90	{
91	return $this->executeBypassingTransaction(new EXEC())->getResponse();	20✔
92	}
93
94	/**
95	* {@inheritDoc}
96	*/
97	public function multi()	3✔
98	{
99	return $this->executeBypassingTransaction(new MULTI())->getResponse();	3✔
100	}
101
102	/**
103	* {@inheritDoc}
104	*/
105	public function watch(array $keys)	7✔
106	{
107	$watch = new WATCH();	7✔
108	$watch->setArguments($keys);	7✔
109
110	return $this->executeBypassingTransaction($watch)->getResponse();	7✔
111	}
112
113	/**
114	* {@inheritDoc}
115	* @throws Throwable
116	*/
117	public function unwatch()	2✔
118	{
119	$retry = $this->connection->getParameters()->retry;	2✔
120
121	return $retry->callWithRetry(	2✔
122	function () {	2✔
123	return $this->connection->executeCommand(new UNWATCH());	2✔
124	}, function (CommunicationException $e) {	2✔
125	$this->onFailCallback($e);	1✔
126	}	2✔
127	);	2✔
128	}
129
130	/**
131	* {@inheritDoc}
132	*/
133	public function discard()	9✔
134	{
135	return $this->executeBypassingTransaction(new DISCARD())->getResponse();	9✔
136	}
137
138	/**
139	* Executes a Redis command bypassing the transaction logic.
140	*
141	* @param CommandInterface $command
142	* @return BypassTransactionResponse
143	* @throws ServerException\|Throwable
144	*/
145	protected function executeBypassingTransaction(CommandInterface $command): BypassTransactionResponse	30✔
146	{
147	$retry = $this->connection->getParameters()->retry;	30✔
148
149	try {
150	$response = $retry->callWithRetry(	30✔
151	function () use ($command) {	30✔
152	return $this->connection->executeCommand($command);	30✔
153	}, function (CommunicationException $e) {	30✔
154	$this->onFailCallback($e);	1✔
155	}	30✔
156	);	30✔
157	} catch (ServerException $exception) {	×
158	if (!$this->connection instanceof RelayConnection) {	×
159	throw $exception;	×
160	}
161
162	if (strcasecmp($command->getId(), 'EXEC') != 0) {	×
163	throw $exception;	×
164	}
165
166	if (!strpos($exception->getMessage(), 'RELAY_ERR_REDIS')) {	×
167	throw $exception;	×
168	}
169
170	return new BypassTransactionResponse(null);	×
171	}
172
173	if ($response instanceof ErrorInterface) {	30✔
174	throw new ServerException($response->getMessage());	2✔
175	}
176
177	return new BypassTransactionResponse($response);	30✔
178	}
179
180	/**
181	* Handle communication exception.
182	*
183	* @param CommunicationException $e
184	* @return void
185	*/
186	private function onFailCallback(CommunicationException $e)	2✔
187	{
188	$connection = $e->getConnection();	2✔
189
190	if ($connection instanceof NodeConnectionInterface) {	2✔
191	$connection->disconnect();	2✔
192
193	return;	2✔
194	}
195
NEW 196	if ($e instanceof ConnectionException) {	×
NEW 197	$nodeConnection = $e->getConnection();	×
198
NEW 199	if ($nodeConnection) {	×
NEW 200	$nodeConnection->disconnect();	×
NEW 201	$this->connection->remove($nodeConnection);	×
202	}
203	}
204
NEW 205	if ($e instanceof TimeoutException) {	×
NEW 206	$nodeConnection = $e->getConnection();	×
207
NEW 208	if ($nodeConnection) {	×
NEW 209	$nodeConnection->disconnect();	×
210	}
211	}
212	}
213	}

predis / predis / 20455852140

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

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