tests: add standalone reproduction for the pipeline issue (#628)

The three "at once" bulk methods open a pipeline that is never discarded
when something goes wrong:

- add_multiple_at_once() and set_multiple_at_once() open the pipeline and
  queue commands outside the try block, so an exception raised while
  queueing (e.g. from a redis_cache_expiration filter) escapes with the
  pipeline still open
- all three catch blocks call handle_exception() and return without
  calling discard(), so a failing exec() leaves the pipeline open too

The connection is then stuck in pipeline mode and answers every later
command with the client object itself, which makes wp_cache_get() fatal
in add_to_internal_cache() ("Trying to clone an uncloneable object"),
wp_cache_get_multiple() fatal in array_combine(), and set/delete report
false while the writes actually land in Redis.

Reproduces identically on PhpRedis 6.3.0 and Relay 0.40.0:

    php tests/pipeline-repro.php
    php tests/pipeline-repro.php --client=relay

Author: claude

tests/pipeline-repro.php

@@ -0,0 +1,278 @@
+<?php
+/**
+ * Standalone reproduction for https://github.com/rhubarbgroup/redis-cache/issues/628
+ *
+ * The three "at once" methods open a pipeline that is never discarded when
+ * something goes wrong, which leaves the connection stuck in pipeline mode.
+ * Every command issued afterwards then returns the client object itself
+ * instead of a value.
+ *
+ * Reproduces identically on PhpRedis and Relay -- it is a drop-in bug, not a
+ * client bug.
+ *
+ * Usage (needs a Redis server on 127.0.0.1:6379):
+ *
+ *     php tests/pipeline-repro.php
+ *     php tests/pipeline-repro.php --client=relay
+ *
+ * @package Rhubarb\RedisCache
+ */
+
+// phpcs:disable
+
+$client = 'phpredis';
+
+foreach ( array_slice( $argv, 1 ) as $arg ) {
+    if ( strpos( $arg, '--client=' ) === 0 ) {
+        $client = substr( $arg, 9 );
+    }
+}
+
+// --------------------------------------------------------------------------
+// Minimal WordPress shim, just enough to boot includes/object-cache.php.
+// --------------------------------------------------------------------------
+
+define( 'ABSPATH', __DIR__ . '/' );
+define( 'WP_CONTENT_DIR', __DIR__ );
+define( 'WP_REDIS_CLIENT', $client );
+define( 'WP_REDIS_HOST', getenv( 'REDIS_HOST' ) ?: '127.0.0.1' );
+define( 'WP_REDIS_PORT', (int) ( getenv( 'REDIS_PORT' ) ?: 6379 ) );
+define( 'WP_REDIS_PREFIX', 'issue628:' );
+define( 'WP_REDIS_GRACEFUL', true );
+
+$GLOBALS['table_prefix'] = 'wp_';
+$GLOBALS['blog_id'] = 1;
+$GLOBALS['wp_filter'] = [];
+
+function add_filter( $hook, $callback, $priority = 10, $accepted_args = 1 ) {
+    $GLOBALS['wp_filter'][ $hook ][] = $callback;
+
+    return true;
+}
+
+function add_action( $hook, $callback, $priority = 10, $accepted_args = 1 ) {
+    return add_filter( $hook, $callback, $priority, $accepted_args );
+}
+
+function has_filter( $hook, $callback = false ) {
+    return ! empty( $GLOBALS['wp_filter'][ $hook ] );
+}
+
+function remove_all_filters( $hook ) {
+    unset( $GLOBALS['wp_filter'][ $hook ] );
+}
+
+function apply_filters( $hook, $value, ...$args ) {
+    foreach ( isset( $GLOBALS['wp_filter'][ $hook ] ) ? $GLOBALS['wp_filter'][ $hook ] : [] as $callback ) {
+        $value = $callback( $value, ...$args );
+    }
+
+    return $value;
+}
+
+function do_action( $hook, ...$args ) {
+    foreach ( isset( $GLOBALS['wp_filter'][ $hook ] ) ? $GLOBALS['wp_filter'][ $hook ] : [] as $callback ) {
+        $callback( ...$args );
+    }
+}
+
+require_once dirname( __DIR__ ) . '/includes/object-cache.php';
+
+// --------------------------------------------------------------------------
+// Helpers.
+// --------------------------------------------------------------------------
+
+$redis_property = new ReflectionProperty( WP_Object_Cache::class, 'redis' );
+$redis_property->setAccessible( true );
+
+function boot_cache() {
+    return $GLOBALS['wp_object_cache'] = new WP_Object_Cache();
+}
+
+function client_of( $cache ) {
+    return $GLOBALS['redis_property']->getValue( $cache );
+}
+
+function replace_client( $cache, $client ) {
+    $GLOBALS['redis_property']->setValue( $cache, $client );
+}
+
+/**
+ * A connection stuck in pipeline mode answers every command with itself.
+ */
+function in_pipeline_mode( $client ) {
+    return $client->echo( 'probe' ) === $client;
+}
+
+function describe( $value ) {
+    if ( is_object( $value ) ) {
+        return get_class( $value ) . ' object';
+    }
+
+    if ( is_array( $value ) ) {
+        return '[' . implode( ', ', array_map( 'describe', $value ) ) . ']';
+    }
+
+    return var_export( $value, true );
+}
+
+function line( $label, $value ) {
+    printf( "  %-44s %s\n", $label, describe( $value ) );
+}
+
+function attempt( $label, callable $callback ) {
+    try {
+        line( $label, $callback() );
+    } catch ( Throwable $throwable ) {
+        line( $label, get_class( $throwable ) . ': ' . $throwable->getMessage() );
+    }
+}
+
+/**
+ * Wraps the real client and fails exec() the way a dropped connection or a
+ * read timeout does, while recording whether discard() is ever called.
+ */
+class FailingExecClient
+{
+    public $inner;
+    public $exec_calls = 0;
+    public $discard_calls = 0;
+
+    public function __construct( $inner ) {
+        $this->inner = $inner;
+    }
+
+    public function pipeline() {
+        $this->inner->pipeline();
+
+        return $this;
+    }
+
+    public function exec() {
+        $this->exec_calls++;
+
+        throw new RuntimeException( 'read error on connection to ' . WP_REDIS_HOST );
+    }
+
+    public function discard() {
+        $this->discard_calls++;
+
+        return $this->inner->discard();
+    }
+
+    public function __call( $method, $arguments ) {
+        $result = $this->inner->{$method}( ...$arguments );
+
+        return $result === $this->inner ? $this : $result;
+    }
+}
+
+// --------------------------------------------------------------------------
+
+$cache = boot_cache();
+
+if ( ! $cache->redis_status() ) {
+    fwrite( STDERR, "Could not connect to Redis.\n" );
+    exit( 1 );
+}
+
+printf( "Client: %s\n\n", $cache->diagnostics['client'] );
+
+echo "1. Baseline -- the bulk methods work\n";
+$cache->flush();
+line( 'add_multiple([a, b])', $cache->add_multiple( [ 'a' => 1, 'b' => 2 ], 'g' ) );
+line( 'connection in pipeline mode?', in_pipeline_mode( client_of( $cache ) ) ? 'YES' : 'no' );
+
+echo "\n2. exec() fails mid-pipeline -- discard() is never called\n";
+echo "   (the catch blocks at object-cache.php:1270, 1510 and 2305 return\n";
+echo "   without taking the connection back out of pipeline mode)\n\n";
+
+foreach ( [ 'add_multiple', 'set_multiple', 'delete_multiple' ] as $method ) {
+    $cache = boot_cache();
+    $real = client_of( $cache );
+    $stub = new FailingExecClient( $real );
+    replace_client( $cache, $stub );
+
+    $payload = $method === 'delete_multiple' ? [ 'k1', 'k2' ] : [ 'k1' => 1, 'k2' => 2 ];
+
+    try {
+        $cache->{$method}( $payload, 'g' );
+    } catch ( Throwable $throwable ) {
+        // Swallowed by handle_exception() already.
+    }
+
+    printf(
+        "  %-19s exec() failed: %-3s discard() called: %-3s left in pipeline mode: %s\n",
+        $method . '()',
+        $stub->exec_calls ? 'yes' : 'no',
+        $stub->discard_calls ? 'yes' : 'no',
+        in_pipeline_mode( $real ) ? 'YES' : 'no'
+    );
+
+    $real->exec();
+}
+
+echo "\n3. An exception while queueing -- same result\n";
+echo "   (in add_multiple_at_once() and set_multiple_at_once() the queueing\n";
+echo "   loop sits outside the try block, so nothing catches this at all)\n\n";
+
+$cache = boot_cache();
+$cache->flush();
+$cache->add_multiple( [ 'a' => 1, 'b' => 2 ], 'g' );
+
+add_filter( 'redis_cache_expiration', function ( $expiration, $key ) {
+    if ( $key === 'boom' ) {
+        throw new RuntimeException( 'a third-party redis_cache_expiration filter threw' );
+    }
+
+    return $expiration;
+}, 10, 4 );
+
+attempt( 'add_multiple([ok, boom, later])', function () use ( $cache ) {
+    return $cache->add_multiple( [ 'ok' => 1, 'boom' => 2, 'later' => 3 ], 'g' );
+} );
+
+remove_all_filters( 'redis_cache_expiration' );
+
+line( 'connection in pipeline mode?', in_pipeline_mode( client_of( $cache ) ) ? 'YES' : 'no' );
+
+echo "\n4. Every cache operation after that point is wrong\n";
+line( 'redis_status()', $cache->redis_status() );
+$cache->flush_runtime();
+attempt( 'get("a", "g")          expected 1', function () use ( $cache ) {
+    return $cache->get( 'a', 'g' );
+} );
+$cache->flush_runtime();
+attempt( 'get_multiple([a, b], "g")', function () use ( $cache ) {
+    return $cache->get_multiple( [ 'a', 'b' ], 'g' );
+} );
+attempt( 'set("c", 3, "g")       expected true', function () use ( $cache ) {
+    return $cache->set( 'c', 3, 'g' );
+} );
+attempt( 'delete("a", "g")       expected true', function () use ( $cache ) {
+    return $cache->delete( 'a', 'g' );
+} );
+attempt( 'add_multiple([x, y], "g")', function () use ( $cache ) {
+    return $cache->add_multiple( [ 'x' => 1, 'y' => 2 ], 'g' );
+} );
+
+echo "\n   ...while the writes that reported false did reach Redis anyway:\n";
+
+$observer = boot_cache();
+line( 'keys issue628:*', client_of( $observer )->keys( 'issue628:*' ) );
+
+echo "\n5. Does the wedged connection survive into the next request?\n";
+echo "   No -- on both clients the next request gets a clean connection, even\n";
+echo "   though Relay's connect() is persistent by default. So the damage is\n";
+echo "   confined to the request that triggered it.\n\n";
+
+$first = boot_cache();
+client_of( $first )->pipeline();
+client_of( $first )->set( 'issue628:leak', 1 );
+unset( $first );
+
+$second = boot_cache();
+line( 'new request, pipeline mode?', in_pipeline_mode( client_of( $second ) ) ? 'YES -- still broken' : 'no -- recovered' );
+attempt( 'new request, get("a", "g")', function () use ( $second ) {
+    return $second->get( 'a', 'g' );
+} );