From 307d7e2b6a61cfe4e36e085253f2324e874eddaa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Till=20Kr=C3=BCss?= Date: Fri, 9 Oct 2026 16:57:05 -0700 Subject: [PATCH] Fix connection cleanup after interrupted pipelines (#1741) Disconnect interrupted pipeline connections before they can be reused, discard remaining commands, and reset running state after PHP errors. Preserve the original failure when connection cleanup also fails. Clear the cached connection resource when Relay is closed so the next command reconnects and replays connection initialization. This fixes client reuse after an interrupted pipeline disconnects the connection. --- CHANGELOG.md | 1 + src/Connection/RelayConnection.php | 2 + src/Pipeline/Pipeline.php | 31 ++- .../Predis/Connection/RelayConnectionTest.php | 17 ++ tests/Predis/Pipeline/PipelineCleanupTest.php | 179 ++++++++++++++++++ 5 files changed, 220 insertions(+), 10 deletions(-) create mode 100644 tests/Predis/Pipeline/PipelineCleanupTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a873da6..bbf95346 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - Added support for numbered databases in Valkey clusters (#1696) ### Changed ### Fixed +- Fixed interrupted pipelines leaving unread replies on reused connections; failed pipelines now discard remaining commands and reset execution state (#1309) ## v3.7.0 (2026-10-08) ### Added diff --git a/src/Connection/RelayConnection.php b/src/Connection/RelayConnection.php index 8e287b3c..5b8077f7 100644 --- a/src/Connection/RelayConnection.php +++ b/src/Connection/RelayConnection.php @@ -114,6 +114,8 @@ class RelayConnection extends AbstractConnection if ($this->client->isConnected()) { $this->client->close(); } + + parent::disconnect(); } /** diff --git a/src/Pipeline/Pipeline.php b/src/Pipeline/Pipeline.php index ab3b946b..83f4fe38 100644 --- a/src/Pipeline/Pipeline.php +++ b/src/Pipeline/Pipeline.php @@ -12,7 +12,6 @@ namespace Predis\Pipeline; -use Exception; use InvalidArgumentException; use Predis\ClientContextInterface; use Predis\ClientException; @@ -308,7 +307,22 @@ class Pipeline implements ClientContextInterface public function flushPipeline($send = true) { if ($send && !$this->pipeline->isEmpty()) { - $responses = $this->executePipeline($this->getConnection(), $this->pipeline); + $connection = $this->getConnection(); + + try { + $responses = $this->executePipeline($connection, $this->pipeline); + } catch (Throwable $exception) { + $this->pipeline = new SplQueue(); + + try { + if ($connection->isConnected()) { + $connection->disconnect(); + } + } finally { + throw $exception; + } + } + $this->responses = array_merge($this->responses, $responses); } else { $this->pipeline = new SplQueue(); @@ -339,7 +353,7 @@ class Pipeline implements ClientContextInterface * @param mixed $callable Optional callback for execution. * * @return array - * @throws Exception + * @throws Throwable * @throws InvalidArgumentException */ public function execute($callable = null) @@ -348,7 +362,6 @@ class Pipeline implements ClientContextInterface throw new InvalidArgumentException('The argument must be a callable object.'); } - $exception = null; $this->setRunning(true); try { @@ -357,14 +370,12 @@ class Pipeline implements ClientContextInterface } $this->flushPipeline(); - } catch (Exception $exception) { - // NOOP - } + } catch (Throwable $exception) { + $this->pipeline = new SplQueue(); - $this->setRunning(false); - - if ($exception) { throw $exception; + } finally { + $this->setRunning(false); } return $this->responses; diff --git a/tests/Predis/Connection/RelayConnectionTest.php b/tests/Predis/Connection/RelayConnectionTest.php index 9edde479..814e72f8 100644 --- a/tests/Predis/Connection/RelayConnectionTest.php +++ b/tests/Predis/Connection/RelayConnectionTest.php @@ -13,6 +13,7 @@ namespace Predis\Connection; use PHPUnit\Framework\MockObject\MockObject; +use Predis\Client; use Predis\ClientException; use Predis\Command\RawCommand; use Predis\NotSupportedException; @@ -455,6 +456,22 @@ class RelayConnectionTest extends PredisTestCase // ---- INTEGRATION TESTS --------------------------------------------- // // ******************************************************************** // + /** + * @group connected + */ + public function testClientReconnectsAfterDisconnect(): void + { + $client = new Client($this->getParameters(), ['connections' => 'relay']); + + $this->assertEquals('PONG', $client->ping()); + $client->disconnect(); + $this->assertFalse($client->isConnected()); + + $this->assertEquals('PONG', $client->ping()); + $this->assertTrue($client->isConnected()); + $client->disconnect(); + } + /** * @group connected */ diff --git a/tests/Predis/Pipeline/PipelineCleanupTest.php b/tests/Predis/Pipeline/PipelineCleanupTest.php new file mode 100644 index 00000000..37b8cdea --- /dev/null +++ b/tests/Predis/Pipeline/PipelineCleanupTest.php @@ -0,0 +1,179 @@ +getMockBuilder($connectionClass)->getMock(); + $connection->method('getParameters')->willReturn(new Parameters()); + $connection->method('isConnected')->willReturn(true); + $connection->expects($this->once())->method('disconnect'); + $connection->expects($this->once())->method($method)->willThrowException($failure); + + $pipeline = new Pipeline(new Client($connection)); + $pipeline->echo('first'); + $pipeline->exists('missing'); + + try { + $pipeline->execute(); + $this->fail('Expected pipeline execution to fail.'); + } catch (Throwable $exception) { + $this->assertSame($failure, $exception); + } + + // Reusing the context must not replay commands already sent to Redis. + $this->assertSame([], $pipeline->execute()); + } + + public function providePipelineFailures(): array + { + $failures = []; + + foreach ([RuntimeException::class, TypeError::class] as $failureClass) { + foreach (['write', 'readResponse'] as $method) { + $failures[] = [NodeConnectionInterface::class, $method, new $failureClass('Interrupted pipeline')]; + } + + $failures[] = [ClusterInterface::class, 'getConnectionByCommand', new $failureClass('Interrupted pipeline')]; + } + + return $failures; + } + + /** + * @group disconnected + */ + public function testPreservesOriginalFailureWhenDisconnectFails(): void + { + $failure = new RuntimeException('Interrupted pipeline'); + $connection = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $connection->method('getParameters')->willReturn(new Parameters()); + $connection->method('isConnected')->willReturn(true); + $connection->method('write')->willThrowException($failure); + $connection->expects($this->once())->method('disconnect') + ->willThrowException(new RuntimeException('Disconnect failed')); + + $pipeline = new Pipeline(new Client($connection)); + $pipeline->ping(); + + try { + $pipeline->flushPipeline(); + $this->fail('Expected pipeline execution to fail.'); + } catch (Throwable $exception) { + $this->assertSame($failure, $exception); + } + } + + /** + * @group disconnected + * @dataProvider provideCallbackFailures + */ + public function testDiscardsQueuedCommandsAndResetsRunningStateAfterCallbackFailure(Throwable $failure): void + { + $connection = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $connection->method('getParameters')->willReturn(new Parameters()); + $connection->expects($this->never())->method('disconnect'); + $connection->expects($this->never())->method('write'); + $connection->expects($this->never())->method('readResponse'); + $pipeline = new Pipeline(new Client($connection)); + + try { + $pipeline->execute(static function (Pipeline $pipe) use ($failure) { + $pipe->set('aborted-write', 'value'); + throw $failure; + }); + $this->fail('Expected callback to fail.'); + } catch (Throwable $exception) { + $this->assertSame($failure, $exception); + } + + $this->assertSame([], $pipeline->execute()); + } + + public function provideCallbackFailures(): array + { + return [ + 'PHP error' => [new TypeError('Callback failed')], + 'exception' => [new RuntimeException('Callback failed')], + ]; + } + + /** + * @group connected + * @requiresRedisVersion >= 2.0.0 + * @dataProvider providePipelineConnections + */ + public function testClientReceivesCorrectRepliesAfterParserFailure(string $pipelineClass, bool $persistent): void + { + $parameters = $this->getParameters(['persistent' => $persistent]); + $client = new Client($parameters, ['connections' => 'default']); + $prefix = 'pipeline-cleanup:' . bin2hex(random_bytes(8)) . ':'; + $hash = $prefix . 'hash'; + $sorted = $prefix . 'sorted'; + $client->hmset($hash, ['field' => 'value']); + $client->zadd($sorted, ['member' => 1]); + $failure = new TypeError('Response parser failed'); + + $command = $this->getMockBuilder(ECHO_::class)->onlyMethods(['parseResponse'])->getMock(); + $command->setArguments(['first']); + $command->expects($this->once())->method('parseResponse')->willThrowException($failure); + + $pipeline = new $pipelineClass($client); + $pipeline->executeCommand($command); + $pipeline->exists($prefix . 'missing'); + + try { + $pipeline->execute(); + $this->fail('Expected response parsing to fail.'); + } catch (Throwable $exception) { + $this->assertSame($failure, $exception); + } + + $this->assertFalse($client->isConnected()); + $this->assertSame(['field' => 'value'], $client->hgetall($hash)); + $this->assertSame(['member' => '1'], $client->zrange($sorted, 0, -1, ['withscores' => true])); + $pipeline->echo('next'); + $this->assertSame(['next'], $pipeline->execute()); + $client->del([$hash, $sorted]); + $client->disconnect(); + } + + public function providePipelineConnections(): array + { + return [ + [Pipeline::class, false], + [Pipeline::class, true], + [Atomic::class, false], + [Atomic::class, true], + ]; + } +}