Add cross-shard pipelines and transactional pipeline blocks to RedisCluster - #2798

Summary

This PR adds native pipelining to RedisCluster while keeping the public API and mode transitions aligned with standalone Redis:

• pipeline() and multi(Redis::PIPELINE) start the same non-atomic pipeline mode.
• Non-atomic pipelines may contain commands for different hash slots and shards.
• pipeline()->multi() creates a Redis transaction inside the pipeline. Each such transaction is restricted to one hash slot.
• Existing multi() behavior is preserved: PhpRedis may coordinate transactions on multiple cluster nodes, with atomicity provided independently by each node rather than globally across shards.
• Distributed MGET, MSET, MSETNX, DEL, and UNLINK operations continue to be split by slot and are folded back into one logical result in the original command order.

This supersedes the original single-slot-only implementation in this PR and addresses the design questions raised in #1910, particularly Michael's failure-handling, MULTI, and multi-key command concerns.

API and atomicity model

For example:

$pipe = $redis->pipeline();
$pipe->set('{a}one', '1')
     ->set('{b}two', '2');
$replies = $pipe->exec();

multi(Redis::PIPELINE) is an alias for the same non-atomic mode:

$pipe = $redis->multi(Redis::PIPELINE);
$pipe->set('{a}one', '1')
     ->get('{a}one');
$replies = $pipe->exec();

A transaction inside a pipeline uses the standalone PhpRedis nesting model. The inner exec() closes and queues the MULTI block; the outer exec() sends the pipeline:

$pipe = $redis->pipeline();
$pipe->multi()
     ->set('{account:1}balance', '100')
     ->get('{account:1}balance')
     ->exec();

$replies = $pipe->exec(); // [[true, '100']]

Keys within that MULTI block must resolve to the same slot, including after prefixing and hash-tag processing. This is a single Redis-node transaction; resharding may still make it fail rather than providing a cross-node atomicity guarantee.

Previously, multi(Redis::PIPELINE) warned and fell back to regular MULTI. It now starts a non-atomic pipeline, matching standalone Redis. Starting a pipeline from an active MULTI is a fatal error. An uncovered slot or a transaction slot violation discards the pending pipeline and restores ATOMIC mode; subsequent commands execute immediately.

Failure and socket-safety model

The implementation is intentionally conservative because a cross-shard pipeline can be partially delivered or partially executed:

• Queueing performs no socket I/O. Commands are buffered per resolved destination socket.
• exec() writes one complete buffer per participating socket and rejects short/partial writes.
• Every queued reply is bound to the exact socket that received the command. Reply processing does not look the socket up again through a potentially changed slot map.
• Replies are folded into the original PHP command order even though commands are grouped by node for transport.
• Normal Redis command errors are represented as false for that result while all remaining replies are consumed, leaving the socket clean.
• On a write failure, truncated reply, invalid framing, MOVED/ASK, or another pipeline-level failure, all participating sockets are disconnected before state is reset. This prevents unread replies or a partially written request from contaminating a later command or a persistent connection.
• Pipelines are not automatically replayed after an ambiguous failure. Some commands may already have executed, so retrying the whole batch could duplicate side effects.
• Queued command contexts have explicit destructors and ownership-transfer state. exec(), discard(), error paths, and object destruction release each context exactly once.
• Reentrant commands on the same RedisCluster object are rejected while pipeline replies are being decoded.

Failure recovery refreshes routing before the next pipeline starts, without replaying failed commands. At least one configured seed must remain reachable.

The shared remap fixes also protect legacy atomic commands: cached seeds retain authentication, rebuilt slot maps clear stale entries, and uncovered slots after MOVED raise an exception.

Commands that cannot be safely represented by this queue are rejected before any pipeline write. This includes directed-node commands, KEYS/scan-style operations, Pub/Sub subscription operations, WATCH/UNWATCH, and directed WAIT/WAITAOF calls.

Distributed multi-key commands

For MGET, MSET, MSETNX, DEL, and UNLINK, the RedisCluster abstraction remains consistent with its existing immediate-command behavior:

1. Keys are prefixed and hashed normally.
2. The logical command is split into per-slot commands.
3. Each fragment is added to the appropriate node buffer.
4. Fragment replies are combined into one logical result in key/argument order.

In a non-atomic pipeline these commands may span shards. Inside a pipelined MULTI block they are accepted only when every key maps to the block's single slot. MSETNX therefore retains RedisCluster's existing per-slot result semantics when it is distributed; it does not become globally atomic across shards.

Testing

The updated test coverage focuses on the failure modes and ownership concerns from #1910:

• Cross-slot and interleaved cross-node pipelines with stable reply ordering.
• Equivalence of pipeline() and multi(Redis::PIPELINE).
• Single-slot pipelined MULTI blocks, multiple blocks on different shards, empty blocks, and hash tags.
• Prefix and serializer behavior, missing keys, distributed multi-key commands, and command errors.
• Wrong-type replies and ACL-denied MSET, MSETNX, DEL, and UNLINK fragments without leaving unread responses.
• WATCH aborts, rejected commands, invalid mode transitions, and reentrant unserialization callbacks.
• Cleanup/leak-oriented lifecycle cases: distributed contexts followed by exec(), discard(), pipeline abort, and object destruction.
• Regression tests that never enter pipeline mode, covering existing immediate commands, distributed multi(), response ordering, error handling, DISCARD, and subsequent socket reuse.
• A non-pipeline regression test covers MOVED followed by a partial slot map, with persistent connections enabled and disabled, and verifies subsequent reads from covered slots.
• Failed-socket reconnect recovery through both pipeline entry APIs, with and without persistent connections, and nested EXEC return-type validation.

Additional validation:

• Full RedisCluster suite, including serializer, compression, and session coverage: passed.
• Full standalone Redis suite: passed.
• Pipeline-disabled regression tests were run against both this branch and a separately built origin/develop module: identical results, all passed.
• Manual resharding/redirection, node failure, and CLUSTERDOWN scenarios were exercised to verify failure returns, state reset, and clean socket recovery.
• Build and generated arginfo consistency checks, git diff --check, and repeated post-failure commands: passed.