From ac3142abfe501c2e28c3e31b064ffb142682507a Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Mon, 10 Aug 2026 10:54:34 +0300 Subject: [PATCH] Added experimental support for HIMPORT bulk hash import feature (#1712) * Added support for HIMPORT bulk hash import feature * Fixed pre-server execution registry miutating, added pre-loaded registry feature * Refactored retry logic to use existing configuration * Marked feature as experimental * Remvoed non-ASCII chars * Ensure IteratorAggregator interface is enforced * Fixed bug with updateCachableException adds duplicate entries * Added explicit reconnect for Relay tests * Testing with custom image * Removed redundant comment * Fixed exception handling * Removed non-ASCII char * Fixed merged conflict bugs --- .github/wordlist.txt | 4 + .github/workflows/tests.yml | 2 +- CHANGELOG.md | 1 + README.md | 78 ++ src/ClientContextInterface.php | 2 + src/ClientInterface.php | 2 + src/Cluster/ClusterStrategy.php | 26 + src/Command/Container/HIMPORT.php | 484 ++++++++++ src/Command/Redis/HIMPORT.php | 64 ++ src/Configuration/Option/Himport.php | 102 ++ src/Configuration/Options.php | 1 + src/Configuration/OptionsInterface.php | 2 + src/Connection/AbstractConnection.php | 55 ++ src/Connection/RelayConnection.php | 26 +- src/Connection/StreamConnection.php | 35 +- src/Himport/FieldsetNotPreparedException.php | 31 + src/Himport/FieldsetRegistry.php | 89 ++ src/Himport/HimportOptions.php | 58 ++ src/Retry/Retry.php | 10 +- tests/Predis/Cluster/PredisStrategyTest.php | 24 + tests/Predis/Cluster/RedisStrategyTest.php | 24 + .../Predis/Command/Container/HIMPORT_Test.php | 530 ++++++++++ tests/Predis/Command/Redis/HIMPORT_Test.php | 910 ++++++++++++++++++ .../Configuration/Option/HimportTest.php | 151 +++ .../Replication/SentinelReplicationTest.php | 71 ++ .../Connection/StreamConnectionTest.php | 105 ++ tests/Predis/Retry/RetryTest.php | 21 + 27 files changed, 2897 insertions(+), 11 deletions(-) create mode 100644 src/Command/Container/HIMPORT.php create mode 100644 src/Command/Redis/HIMPORT.php create mode 100644 src/Configuration/Option/Himport.php create mode 100644 src/Himport/FieldsetNotPreparedException.php create mode 100644 src/Himport/FieldsetRegistry.php create mode 100644 src/Himport/HimportOptions.php create mode 100644 tests/Predis/Command/Container/HIMPORT_Test.php create mode 100644 tests/Predis/Command/Redis/HIMPORT_Test.php create mode 100644 tests/Predis/Configuration/Option/HimportTest.php diff --git a/.github/wordlist.txt b/.github/wordlist.txt index a8cc67d4..340526f8 100644 --- a/.github/wordlist.txt +++ b/.github/wordlist.txt @@ -56,3 +56,7 @@ invalidations Sharded sharded Valkey +HIMPORT +fieldset +fieldsets +failover diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 0ad2b8bc..4cd07513 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -46,7 +46,7 @@ jobs: run: | # Mapping of original redis versions to client test containers declare -A redis_clients_version_mapping=( - ["8.10"]="8.10-rc2" + ["8.10"]="custom-30445126297-debian" ["8.8"]="8.8.0" ["8.6"]="8.6.1" ["8.4"]="8.4.0" diff --git a/CHANGELOG.md b/CHANGELOG.md index 8afe6ec8..dc9be40a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ - Added support for `FT.ALIASLIST` command - Added stream commands to ClusterStrategy - Added vector sets commands to ClusterStrategy +- Added experimental support for `HIMPORT` bulk hash import feature (API may change in a future release) - Added `OBJECT` and hash field expiration commands to ClusterStrategy ### Fixed diff --git a/README.md b/README.md index a91a4ed6..2c72168b 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ More details about this project can be found on the [frequently asked questions] - Command pipelining on both single nodes and clusters (client-side sharding only). - Abstraction for Redis transactions (Redis >= 2.0) and CAS operations (Redis >= 2.2). - Abstraction for Lua scripting (Redis >= 2.6) and automatic switching between `EVALSHA` or `EVAL`. +- Abstraction for Hinted Hash Templates (`HIMPORT`, Redis >= 8.10) with automatic per-connection fieldset replay. - Abstraction for `SCAN`, `SSCAN`, `ZSCAN` and `HSCAN` (Redis >= 2.8) based on PHP iterators. - Connections are established lazily by the client upon the first command and can be persisted. - Connections can be established via TCP/IP (also TLS/SSL-encrypted) or UNIX domain sockets. @@ -434,6 +435,83 @@ $response = $redis->transaction(function (MultiExec $tx) { ``` +### Hinted Hash Templates (HIMPORT) ### + +> **Experimental:** this feature is experimental and its API (the `himport` container and option) may +> change in a future release. + +`HIMPORT` (Redis >= 8.10) speeds up loading many hashes that share the same field names. The field +names are sent once with `HIMPORT PREPARE`, registering them under a fieldset name, and hashes are +then created with `HIMPORT SET` by sending only the values. The server stores such hashes in a +memory-efficient encoding where the field names are kept only once; the resulting keys are ordinary +hashes that work with every regular hash command. + +Predis exposes the command family through the `himport` container: + +```php +$client->himport->prepare('users', ['name', 'email', 'age']); + +$client->himport->set('user:1', 'users', ['alice', 'alice@example.com', '25']); +$client->himport->set('user:2', 'users', ['bob', 'bob@example.com', '30']); + +$client->himport->discard('users'); // 1 +$client->himport->discardAll(); // number of fieldsets removed +``` + +Values are paired by position with the fields supplied to `prepare()`, in the caller's order — Predis +never reorders them. Hash enumeration order (e.g. `HGETALL`) is not guaranteed to match the `PREPARE` +order, only the value-to-field pairing is. + +A fieldset is **server-side session state that lives on a single physical connection** and is lost when +that connection is dropped (reconnect, `RESET`, cluster failover). To keep this transparent, Predis +tracks the fieldsets prepared through the container and: + +- replays each `PREPARE` automatically when a connection is re-established (at most once per physical + connection); this always happens and does not depend on retries; +- if a `HIMPORT SET` still reports `no such fieldset` (for example on a connection created after a + cluster redirection), re-prepares the fieldset on the executing connection and retries the write. + +The re-prepare-and-retry step **reuses the client's configured retry policy** — it is not a separate +mechanism. It therefore only happens when retries are enabled (via the `retry` connection parameter), +and never more than the configured number of attempts; with retries disabled the `no such fieldset` +error propagates unchanged. It can additionally be turned off, even when retries are enabled, with the +`himport` option: + +```php +$client = new Predis\Client( + $parameters + ['retry' => new Predis\Retry\Retry(new Predis\Retry\Strategy\ExponentialBackoff(), 3)], + ['himport' => ['auto_prepare' => false]] // opt out of HIMPORT re-prepare specifically +); +``` + +Fieldsets can also be declared up front through the `himport` option. Fieldsets declared this way are +prepared on demand the first time a `HIMPORT SET` references them on a connection, so the application +never has to call `prepare()` for them (this uses the re-prepare-and-retry path above, so it requires +retries to be enabled): + +```php +$client = new Predis\Client($parameters + ['retry' => new Predis\Retry\Retry(new Predis\Retry\Strategy\ExponentialBackoff(), 3)], [ + 'himport' => [ + 'fieldsets' => [ + 'users' => ['name', 'email', 'age'], + ], + ], +]); + +// No prepare() call needed — "users" is known from configuration: +$client->himport->set('user:1', 'users', ['alice', 'alice@example.com', '25']); +``` + +On a cluster, `prepare()`, `discard()` and `discardAll()` fan out to every master shard, while `set()` +is routed by the hash slot of its key like any other write. This ensures a `HIMPORT SET` succeeds on +whichever shard owns its key. + +The raw command form (`$client->himport('PREPARE', 'users', 'name', 'email')`) is also available and is +the one to use inside pipelines and transactions; it performs no client-side tracking or recovery, so +`PREPARE` and the dependent `SET` commands must run on the same connection (which pipelines and +transactions guarantee). + + ### Adding new commands ### While we try to update Predis to stay up to date with all the commands available in Redis, you might diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php index 1cb27a58..704beb15 100644 --- a/src/ClientContextInterface.php +++ b/src/ClientContextInterface.php @@ -44,6 +44,7 @@ use Predis\Command\CommandInterface; use Predis\Command\Container\ACL; use Predis\Command\Container\CLIENT; use Predis\Command\Container\FUNCTIONS; +use Predis\Command\Container\HIMPORT; use Predis\Command\Container\HOTKEYS; use Predis\Command\Container\Json\JSONDEBUG; use Predis\Command\Container\Search\FTCONFIG; @@ -431,6 +432,7 @@ use Predis\Command\Redis\VADD; * * Container commands * @property CLIENT $client + * @property HIMPORT $himport * @property HOTKEYS $hotkeys * @property FUNCTIONS $function * @property FTCONFIG $ftconfig diff --git a/src/ClientInterface.php b/src/ClientInterface.php index 49afff2b..fc61dcc0 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -44,6 +44,7 @@ use Predis\Command\CommandInterface; use Predis\Command\Container\ACL; use Predis\Command\Container\CLIENT; use Predis\Command\Container\FUNCTIONS; +use Predis\Command\Container\HIMPORT; use Predis\Command\Container\HOTKEYS; use Predis\Command\Container\Json\JSONDEBUG; use Predis\Command\Container\Search\FTCONFIG; @@ -446,6 +447,7 @@ use Predis\Response\Status; * * Container commands * @property CLIENT $client + * @property HIMPORT $himport * @property HOTKEYS $hotkeys * @property FUNCTIONS $function * @property FTCONFIG $ftconfig diff --git a/src/Cluster/ClusterStrategy.php b/src/Cluster/ClusterStrategy.php index d8b2c65f..547deae5 100644 --- a/src/Cluster/ClusterStrategy.php +++ b/src/Cluster/ClusterStrategy.php @@ -180,6 +180,7 @@ abstract class ClusterStrategy implements StrategyInterface 'HVALS' => $getKeyFromFirstArgument, 'HSCAN' => $getKeyFromFirstArgument, 'HSTRLEN' => $getKeyFromFirstArgument, + 'HIMPORT' => [$this, 'getKeyFromHimportCommands'], 'HEXPIRE' => $getKeyFromFirstArgument, 'HEXPIREAT' => $getKeyFromFirstArgument, 'HPERSIST' => $getKeyFromFirstArgument, @@ -481,6 +482,31 @@ abstract class ClusterStrategy implements StrategyInterface return $arguments[1]; } + /** + * Extracts the key from HIMPORT commands. + * + * Only HIMPORT SET operates on a key (at position 1, after the subcommand) + * and is routed by its hash slot. HIMPORT PREPARE/DISCARD/DISCARDALL are + * connection-session commands that the server advertises as all-shards + * commands; they have no key, so this returns null and the cluster + * connection rejects them. The dedicated container command is the sanctioned + * path for fanning those out across the master shards. + * + * @param CommandInterface $command Command instance. + * + * @return string|null + */ + protected function getKeyFromHimportCommands(CommandInterface $command) + { + $arguments = $command->getArguments(); + + if (isset($arguments[0], $arguments[1]) && strtoupper($arguments[0]) === 'SET') { + return $arguments[1]; + } + + return null; + } + /** * Extracts the key from GEORADIUS and GEORADIUSBYMEMBER commands. * diff --git a/src/Command/Container/HIMPORT.php b/src/Command/Container/HIMPORT.php new file mode 100644 index 00000000..7f93bf85 --- /dev/null +++ b/src/Command/Container/HIMPORT.php @@ -0,0 +1,484 @@ + false]`. + * + * The recovery layer only ever acts on fieldsets declared through this container + * (or the `himport` option). The raw command form ($client->himport('SET', ...)), + * pipelines and transactions stay fully explicit: they never recover and + * propagate server errors as-is. + * + * @experimental This API is experimental and may change in a future release. + * + * @method Status prepare(string $fieldset, array $fields) + * @method Status set(string $key, string $fieldset, array $values) + * @method int discard(string $fieldset) + * @method int discardAll() + */ +class HIMPORT extends AbstractContainer +{ + private const SUBCOMMAND_PREPARE = 'PREPARE'; + private const SUBCOMMAND_SET = 'SET'; + private const SUBCOMMAND_DISCARD = 'DISCARD'; + private const SUBCOMMAND_DISCARDALL = 'DISCARDALL'; + + private const NO_SUCH_FIELDSET = 'no such fieldset'; + private const SESSION_KEY_PREFIX = 'himport:'; + + public function getContainerCommandId(): string + { + return 'HIMPORT'; + } + + /** + * Registers an ordered field list under a fieldset name for use by later + * SET calls. On a cluster this is fanned out to every master shard. + * + * @param string $fieldset + * @param array $fields Ordered, non-empty field names (sent verbatim). + * @return Status|ErrorInterface + * @throws ServerException + */ + public function prepare(string $fieldset, array $fields) + { + if (empty($fields)) { + throw new InvalidArgumentException('HIMPORT PREPARE requires a non-empty list of fields.'); + } + + $command = $this->client->createCommand('HIMPORT', [self::SUBCOMMAND_PREPARE, $fieldset, $fields]); + $connection = $this->client->getConnection(); + + if ($connection instanceof ClusterInterface) { + $results = $this->fanOut($connection, $command); + + if (null !== $error = $this->firstError($results)) { + // Do not record a fieldset the server rejected: a poison entry + // would make every later SET retry a PREPARE that cannot succeed. + return $this->surfaceError($error); + } + + // Record and pin only after every master confirmed the PREPARE. + $this->getRegistry()->set($fieldset, $fields); + + foreach ($results as $result) { + $this->pinSessionCommand($result[0], $this->sessionKey($fieldset), $command); + } + + return Status::get('OK'); + } + + $response = $this->client->executeCommand($command); + + if ($response instanceof ErrorInterface) { + return $response; + } + + $this->getRegistry()->set($fieldset, $fields); + $this->pinSessionCommand($this->resolveNode($command), $this->sessionKey($fieldset), $command); + + return $response; + } + + /** + * Creates or overwrites a hash at $key using the field list previously + * prepared under $fieldset on the executing connection. + * + * If the fieldset is missing on the executing connection but known to the + * client, the write is re-prepared and retried through the connection's Retry + * policy: this happens only when retries are configured, and never more than + * the configured number of attempts. With retries disabled the "no such + * fieldset" error propagates unchanged. + * + * @param string $key + * @param string $fieldset + * @param array $values Ordered, non-empty values paired positionally with the prepared fields. + * @return Status|ErrorInterface + * @throws ServerException + */ + public function set(string $key, string $fieldset, array $values) + { + if (empty($values)) { + throw new InvalidArgumentException('HIMPORT SET requires a non-empty list of values.'); + } + + $command = $this->client->createCommand('HIMPORT', [self::SUBCOMMAND_SET, $key, $fieldset, $values]); + $retry = $this->getConfiguredRetry(); + + try { + if (null === $retry) { + // No retry policy available: run once, no recovery. + return $this->executeSet($command, $fieldset); + } + + // Reuse the client-configured Retry (its enabled/disabled state, count + // and backoff). Register our retryable condition on it, the same way + // the cluster and sentinel connections register theirs. + $retry->updateCatchableExceptions([FieldsetNotPreparedException::class]); + + return $retry->callWithRetry( + function () use ($command, $fieldset) { + return $this->executeSet($command, $fieldset); + }, + function (Throwable $exception) use ($command, $fieldset) { + // Only HIMPORT's own "no such fieldset" is recovered and + // retried here; anything else the shared policy might catch + // (connection errors, other server errors) is left to its + // owner and propagated immediately. + if (!$exception instanceof FieldsetNotPreparedException) { + throw $exception; + } + + $this->reprepare($command, $fieldset); + } + ); + } catch (FieldsetNotPreparedException $exception) { + // Retries were disabled or exhausted. Preserve the client's error + // semantics: return the error response when exceptions are off. + if (!$this->client->getOptions()->exceptions) { + return new Error($exception->getMessage()); + } + + throw $exception; + } + } + + /** + * Removes a fieldset from the connection session. On a cluster this is + * fanned out to every master shard. + * + * @param string $fieldset + * @return int|ErrorInterface Number of removals (1 if present, 0 otherwise). + * @throws ServerException + */ + public function discard(string $fieldset) + { + $command = $this->client->createCommand('HIMPORT', [self::SUBCOMMAND_DISCARD, $fieldset]); + $connection = $this->client->getConnection(); + + if ($connection instanceof ClusterInterface) { + $results = $this->fanOut($connection, $command); + + if (null !== $error = $this->firstError($results)) { + // Leave the registry and pins untouched so the fieldset stays + // consistently known and the discard can be retried. + return $this->surfaceError($error); + } + + // Drop registry entry and pins together, only after every master + // confirmed the discard, so the two never disagree. + $this->getRegistry()->remove($fieldset); + $removed = 0; + + foreach ($results as $result) { + $this->unpinSessionCommand($result[0], $this->sessionKey($fieldset)); + $removed = max($removed, (int) $result[1]); + } + + return $removed; + } + + $response = $this->client->executeCommand($command); + + if ($response instanceof ErrorInterface) { + return $response; + } + + $this->getRegistry()->remove($fieldset); + $this->unpinSessionCommand($this->resolveNode($command), $this->sessionKey($fieldset)); + + return $response; + } + + /** + * Removes all fieldsets from the connection session. On a cluster this is + * fanned out to every master shard. + * + * @return int|ErrorInterface Number of fieldsets removed. + * @throws ServerException + */ + public function discardAll() + { + $command = $this->client->createCommand('HIMPORT', [self::SUBCOMMAND_DISCARDALL]); + $connection = $this->client->getConnection(); + + if ($connection instanceof ClusterInterface) { + $results = $this->fanOut($connection, $command); + + if (null !== $error = $this->firstError($results)) { + // Leave client-side state intact so it stays consistent and the + // discard can be retried. + return $this->surfaceError($error); + } + + $this->getRegistry()->clear(); + $removed = 0; + + foreach ($results as $result) { + $this->unpinSessionCommandsByPrefix($result[0], self::SESSION_KEY_PREFIX); + $removed = max($removed, (int) $result[1]); + } + + return $removed; + } + + $response = $this->client->executeCommand($command); + + if ($response instanceof ErrorInterface) { + return $response; + } + + $this->getRegistry()->clear(); + $this->unpinSessionCommandsByPrefix($this->resolveNode($command), self::SESSION_KEY_PREFIX); + + return $response; + } + + /** + * Executes the SET, translating a recoverable "no such fieldset" into a + * retryable exception so the Retry policy can drive re-prepare + retry. + * Non-recoverable errors are returned/propagated unchanged. + * + * @param CommandInterface $command + * @param string $fieldset + * @return Status|ErrorInterface + * @throws FieldsetNotPreparedException + * @throws ServerException + */ + private function executeSet(CommandInterface $command, string $fieldset) + { + try { + $response = $this->client->executeCommand($command); + } catch (ServerException $exception) { + if ($this->canRecover($exception->getMessage(), $fieldset)) { + throw new FieldsetNotPreparedException($exception->getMessage()); + } + + throw $exception; + } + + if ($response instanceof ErrorInterface && $this->canRecover($response->getMessage(), $fieldset)) { + throw new FieldsetNotPreparedException($response->getMessage()); + } + + return $response; + } + + /** + * Re-prepares $fieldset on the connection the (next) SET attempt targets. + * A failed re-prepare aborts recovery and is surfaced as the root cause, + * through the same "no such fieldset" path so it honours the client's + * exceptions option (thrown when on, returned as an Error when off). + * + * @param CommandInterface $setCommand + * @param string $fieldset + * @throws FieldsetNotPreparedException + */ + private function reprepare(CommandInterface $setCommand, string $fieldset): void + { + $fields = $this->getRegistry()->get($fieldset); + $prepare = $this->client->createCommand('HIMPORT', [self::SUBCOMMAND_PREPARE, $fieldset, $fields]); + + // Resolve after the SET failed so that, following a MOVED/ASK redirection, + // we re-prepare on the node the retry will actually target. + $node = $this->resolveNode($setCommand); + $response = $node->executeCommand($prepare); + + if ($response instanceof ErrorInterface) { + // Surface the PREPARE error as the root cause via set()'s catch, which + // applies the exceptions option (throw vs. return an Error response). + throw new FieldsetNotPreparedException($response->getMessage()); + } + + $this->pinSessionCommand($node, $this->sessionKey($fieldset), $prepare); + } + + /** + * Returns the Retry policy configured on the executing connection, or null + * when none is available. HIMPORT does not define its own retry behaviour: + * it reuses whatever the client is configured with, so retries happen only + * when (and as many times as) the user enabled them. + * + * @return Retry|null + */ + private function getConfiguredRetry(): ?Retry + { + $parameters = $this->client->getConnection()->getParameters(); + + if (null !== $parameters && $parameters->retry instanceof Retry) { + return $parameters->retry; + } + + return null; + } + + /** + * @param string $message Server error message. + * @param string $fieldset Fieldset referenced by the failed SET. + */ + private function canRecover(string $message, string $fieldset): bool + { + return false !== stripos($message, self::NO_SUCH_FIELDSET) + && $this->getOptions()->isAutoPrepareEnabled() + && $this->getRegistry()->has($fieldset); + } + + /** + * Executes a command on every master shard of a cluster connection. + * + * Fails fast rather than silently no-op'ing: a cluster that cannot be + * enumerated, or that yields no master shards, must never look like a + * successful fan-out to the callers that mutate the registry and pins. + * + * @param ClusterInterface $cluster + * @param CommandInterface $command + * @return array + * @throws NotSupportedException + */ + private function fanOut(ClusterInterface $cluster, CommandInterface $command): array + { + if (!$cluster instanceof IteratorAggregate) { + throw new NotSupportedException(sprintf( + "HIMPORT requires an iterable cluster connection to fan out to master shards; '%s' is not iterable.", + get_class($cluster) + )); + } + + $results = []; + + foreach ($cluster->getIterator() as $node) { + $results[] = [$node, $node->executeCommand($command)]; + } + + if (empty($results)) { + throw new NotSupportedException( + 'HIMPORT could not fan out: the cluster connection reported no master shards.' + ); + } + + return $results; + } + + /** + * @param array $results + * @return ErrorInterface|null + */ + private function firstError(array $results): ?ErrorInterface + { + foreach ($results as $result) { + if ($result[1] instanceof ErrorInterface) { + return $result[1]; + } + } + + return null; + } + + /** + * @param ErrorInterface $error + * @return ErrorInterface + * @throws ServerException + */ + private function surfaceError(ErrorInterface $error) + { + if ($this->client->getOptions()->exceptions) { + throw new ServerException($error->getMessage()); + } + + return $error; + } + + /** + * Resolves the physical connection that executes (or executed) $command. + * + * @param CommandInterface $command + * @return NodeConnectionInterface + */ + private function resolveNode(CommandInterface $command): NodeConnectionInterface + { + $connection = $this->client->getConnection(); + + if ($connection instanceof AggregateConnectionInterface) { + return $connection->getConnectionByCommand($command); + } + + return $connection; + } + + private function pinSessionCommand(NodeConnectionInterface $node, string $key, CommandInterface $command): void + { + if (method_exists($node, 'addSessionCommand')) { + $node->addSessionCommand($key, $command); + } + } + + private function unpinSessionCommand(NodeConnectionInterface $node, string $key): void + { + if (method_exists($node, 'removeSessionCommand')) { + $node->removeSessionCommand($key); + } + } + + private function unpinSessionCommandsByPrefix(NodeConnectionInterface $node, string $prefix): void + { + if (method_exists($node, 'removeSessionCommandsByPrefix')) { + $node->removeSessionCommandsByPrefix($prefix); + } + } + + private function sessionKey(string $fieldset): string + { + return self::SESSION_KEY_PREFIX . $fieldset; + } + + private function getOptions(): HimportOptions + { + return $this->client->getOptions()->himport; + } + + private function getRegistry() + { + return $this->getOptions()->getRegistry(); + } +} diff --git a/src/Command/Redis/HIMPORT.php b/src/Command/Redis/HIMPORT.php new file mode 100644 index 00000000..92a78389 --- /dev/null +++ b/src/Command/Redis/HIMPORT.php @@ -0,0 +1,64 @@ +getArguments(); + + // Only HIMPORT SET carries a key (at index 1, after the subcommand). + // PREPARE/DISCARD/DISCARDALL operate on fieldset names, never keys. + if (isset($arguments[0], $arguments[1]) && strtoupper($arguments[0]) === 'SET') { + $arguments[1] = $prefix . $arguments[1]; + $this->setRawArguments($arguments); + } + } +} diff --git a/src/Configuration/Option/Himport.php b/src/Configuration/Option/Himport.php new file mode 100644 index 00000000..8eaa3949 --- /dev/null +++ b/src/Configuration/Option/Himport.php @@ -0,0 +1,102 @@ + ordered, non-empty field + * list. Pre-declared fieldsets are prepared on demand the + * first time a `HIMPORT SET` references them on a connection, + * so the application does not need to call `prepare()` for + * them (this uses the same auto-prepare mechanism below). + * - `auto_prepare`: optional bool (default true) toggling whether the himport + * container prepares/re-prepares fieldsets on demand and + * recovers from "no such fieldset". + * + * @experimental This option is experimental and its shape may change in a future release. + */ +class Himport implements OptionInterface +{ + /** + * {@inheritdoc} + */ + public function filter(OptionsInterface $options, $value) + { + if ($value instanceof HimportOptions) { + return $value; + } + + if (!is_array($value)) { + throw new InvalidArgumentException( + 'Invalid value for the himport option: expected an array or a Predis\Himport\HimportOptions instance.' + ); + } + + $autoPrepare = array_key_exists('auto_prepare', $value) + ? (bool) $value['auto_prepare'] + : true; + + $registry = null; + + if (array_key_exists('fieldsets', $value)) { + $registry = $this->buildRegistry($value['fieldsets']); + } + + return new HimportOptions($registry, $autoPrepare); + } + + /** + * {@inheritdoc} + */ + public function getDefault(OptionsInterface $options) + { + return new HimportOptions(null, true); + } + + /** + * Builds a fieldset registry pre-seeded from the `fieldsets` option. + * + * @param mixed $fieldsets + * @return FieldsetRegistry + */ + private function buildRegistry($fieldsets): FieldsetRegistry + { + if (!is_array($fieldsets)) { + throw new InvalidArgumentException( + 'The "fieldsets" himport option must be a map of fieldset name => list of field names.' + ); + } + + $registry = new FieldsetRegistry(); + + foreach ($fieldsets as $name => $fields) { + if (!is_array($fields) || empty($fields)) { + throw new InvalidArgumentException( + sprintf('Fieldset "%s" must be a non-empty list of field names.', $name) + ); + } + + $registry->set((string) $name, array_values($fields)); + } + + return $registry; + } +} diff --git a/src/Configuration/Options.php b/src/Configuration/Options.php index 3a051d06..6e98c86a 100644 --- a/src/Configuration/Options.php +++ b/src/Configuration/Options.php @@ -33,6 +33,7 @@ class Options implements OptionsInterface 'prefix' => Option\Prefix::class, 'crc16' => Option\CRC16::class, 'upstream_driver' => Option\UpstreamDriver::class, + 'himport' => Option\Himport::class, ]; /** @var array */ diff --git a/src/Configuration/OptionsInterface.php b/src/Configuration/OptionsInterface.php index 20b9a065..a6040738 100644 --- a/src/Configuration/OptionsInterface.php +++ b/src/Configuration/OptionsInterface.php @@ -15,6 +15,7 @@ namespace Predis\Configuration; use Predis\Command\Processor\ProcessorInterface; use Predis\Connection\FactoryInterface; use Predis\Connection\ParametersInterface; +use Predis\Himport\HimportOptions; /** * @property callable $aggregate Custom aggregate connection initializer @@ -27,6 +28,7 @@ use Predis\Connection\ParametersInterface; * @property callable $replication Aggregate connection initializer for replication * @property int $readTimeout Timeout in milliseconds between read operations on reading from multiple connections. * @property string|string[] $upstream_driver Upstream driver info for CLIENT SETINFO. + * @property HimportOptions $himport HIMPORT fieldset registry and auto-recovery configuration. */ interface OptionsInterface { diff --git a/src/Connection/AbstractConnection.php b/src/Connection/AbstractConnection.php index b61c425d..03629a98 100644 --- a/src/Connection/AbstractConnection.php +++ b/src/Connection/AbstractConnection.php @@ -47,6 +47,17 @@ abstract class AbstractConnection implements NodeConnectionInterface */ protected $initCommands = []; + /** + * Commands that restore server-side session state (e.g. HIMPORT fieldsets) + * on the physical connection. Unlike init commands, these are keyed so they + * can be replaced and removed, are replayed best-effort after init commands + * on every (re)connect, and are intentionally excluded from serialization: a + * deserialized or freshly created connection is a new server session. + * + * @var array + */ + protected $sessionCommands = []; + /** * @param ParametersInterface $parameters Initialization parameters for the connection. */ @@ -127,6 +138,50 @@ abstract class AbstractConnection implements NodeConnectionInterface return $this->initCommands; } + /** + * Registers (or replaces) a session command under the given key so it is + * replayed on the next (re)connect. Keying makes replacement idempotent. + * + * @param string $key Identifier for the session command. + * @param CommandInterface $command Command replayed to restore session state. + */ + public function addSessionCommand(string $key, CommandInterface $command): void + { + $this->sessionCommands[$key] = $command; + } + + /** + * Removes a single session command previously registered under the key. + * + * @param string $key Identifier for the session command. + */ + public function removeSessionCommand(string $key): void + { + unset($this->sessionCommands[$key]); + } + + /** + * Removes every session command whose key starts with the given prefix. + * + * @param string $prefix Key prefix (e.g. "himport:") to match. + */ + public function removeSessionCommandsByPrefix(string $prefix): void + { + foreach (array_keys($this->sessionCommands) as $key) { + if (strpos($key, $prefix) === 0) { + unset($this->sessionCommands[$key]); + } + } + } + + /** + * @return array + */ + public function getSessionCommands(): array + { + return $this->sessionCommands; + } + /** * {@inheritdoc} */ diff --git a/src/Connection/RelayConnection.php b/src/Connection/RelayConnection.php index b48e054d..8e287b3c 100644 --- a/src/Connection/RelayConnection.php +++ b/src/Connection/RelayConnection.php @@ -369,15 +369,25 @@ class RelayConnection extends AbstractConnection */ public function connect() { - if (parent::connect() && $this->initCommands) { - foreach ($this->initCommands as $command) { - $response = $this->executeCommand($command); + if (!parent::connect()) { + return; + } - if ($response instanceof ErrorResponseInterface && ($command->getId() === 'CLIENT')) { - // Do nothing on CLIENT SETINFO command failure - } elseif ($response instanceof ErrorResponseInterface) { - $this->onConnectionError("`{$command->getId()}` failed: {$response->getMessage()}", 0); - } + foreach ($this->initCommands as $command) { + $response = $this->executeCommand($command); + + if ($response instanceof ErrorResponseInterface && ($command->getId() === 'CLIENT')) { + // Do nothing on CLIENT SETINFO command failure + } elseif ($response instanceof ErrorResponseInterface) { + $this->onConnectionError("`{$command->getId()}` failed: {$response->getMessage()}", 0); + } + } + + foreach ($this->sessionCommands as $key => $command) { + $response = $this->executeCommand($command); + + if ($response instanceof ErrorResponseInterface) { + unset($this->sessionCommands[$key]); } } } diff --git a/src/Connection/StreamConnection.php b/src/Connection/StreamConnection.php index 6ed0ebc4..1dd5d830 100644 --- a/src/Connection/StreamConnection.php +++ b/src/Connection/StreamConnection.php @@ -89,7 +89,11 @@ class StreamConnection extends AbstractConnection */ public function connect() { - if (parent::connect() && $this->initCommands) { + if (!parent::connect()) { + return; + } + + if ($this->initCommands) { $responses = $this->sendPipeline($this->initCommands); if ($responses[0][0] instanceof ErrorResponseInterface) { @@ -103,6 +107,35 @@ class StreamConnection extends AbstractConnection $this->handleOnConnectResponse($response[0], $response[1]); } } + + $this->replaySessionCommands(); + } + + /** + * Replays registered session commands after a (re)connect. + * + * Replay is best-effort: a server error for a single command drops only that + * entry and never tears down the connection (contrast with init commands, + * whose failures raise a ConnectionException). Whatever state could not be + * restored is surfaced later as the authoritative server error on the first + * command that depends on it. + * + * @throws CommunicationException + */ + protected function replaySessionCommands(): void + { + if (!$this->sessionCommands) { + return; + } + + $keys = array_keys($this->sessionCommands); + $responses = $this->sendPipeline(array_values($this->sessionCommands)); + + foreach ($responses as $index => $response) { + if ($response[0] instanceof ErrorResponseInterface) { + unset($this->sessionCommands[$keys[$index]]); + } + } } /** diff --git a/src/Himport/FieldsetNotPreparedException.php b/src/Himport/FieldsetNotPreparedException.php new file mode 100644 index 00000000..f8a8e1dd --- /dev/null +++ b/src/Himport/FieldsetNotPreparedException.php @@ -0,0 +1,31 @@ + + */ + private $fieldsets = []; + + /** + * Registers a fieldset, replacing any previous definition (last PREPARE + * wins, matching the server's silent-replace semantics). Field order is + * preserved verbatim. + * + * @param string $name Fieldset name. + * @param array $fields Ordered field names. + */ + public function set(string $name, array $fields): void + { + $this->fieldsets[$name] = array_values($fields); + } + + /** + * @param string $name Fieldset name. + * @return array|null The ordered field list, or null when not registered. + */ + public function get(string $name): ?array + { + return $this->fieldsets[$name] ?? null; + } + + /** + * @param string $name Fieldset name (empty string is a valid name). + * @return bool + */ + public function has(string $name): bool + { + return array_key_exists($name, $this->fieldsets); + } + + /** + * @param string $name Fieldset name to forget. + */ + public function remove(string $name): void + { + unset($this->fieldsets[$name]); + } + + /** + * Forgets every registered fieldset. + */ + public function clear(): void + { + $this->fieldsets = []; + } + + /** + * @return array + */ + public function all(): array + { + return $this->fieldsets; + } +} diff --git a/src/Himport/HimportOptions.php b/src/Himport/HimportOptions.php new file mode 100644 index 00000000..5f72b2d4 --- /dev/null +++ b/src/Himport/HimportOptions.php @@ -0,0 +1,58 @@ +registry = $registry ?: new FieldsetRegistry(); + $this->autoPrepare = $autoPrepare; + } + + /** + * @return FieldsetRegistry + */ + public function getRegistry(): FieldsetRegistry + { + return $this->registry; + } + + /** + * @return bool + */ + public function isAutoPrepareEnabled(): bool + { + return $this->autoPrepare; + } +} diff --git a/src/Retry/Retry.php b/src/Retry/Retry.php index d2a19e16..d93c2a45 100644 --- a/src/Retry/Retry.php +++ b/src/Retry/Retry.php @@ -72,12 +72,20 @@ class Retry /** * Extend catchable exceptions list. * + * Already-registered classes are ignored, so callers may safely re-register + * on every command without the list (and the per-failure lookup) growing + * without bound. + * * @param array $catchableExceptions * @return void */ public function updateCatchableExceptions(array $catchableExceptions): void { - $this->catchableExceptions = array_merge($this->catchableExceptions, $catchableExceptions); + foreach ($catchableExceptions as $catchableException) { + if (!in_array($catchableException, $this->catchableExceptions, true)) { + $this->catchableExceptions[] = $catchableException; + } + } } /** diff --git a/tests/Predis/Cluster/PredisStrategyTest.php b/tests/Predis/Cluster/PredisStrategyTest.php index f3dcd877..5b2d5ded 100644 --- a/tests/Predis/Cluster/PredisStrategyTest.php +++ b/tests/Predis/Cluster/PredisStrategyTest.php @@ -350,6 +350,29 @@ class PredisStrategyTest extends PredisTestCase } } + /** + * @group disconnected + */ + public function testKeysForHimportCommand(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // HIMPORT SET is routed by the hash slot of its key (at position 1). + $command = $commands->create('HIMPORT', ['SET', '{key}:1', 'fieldset', 'v1', 'v2']); + $this->assertNotNull($strategy->getSlot($command)); + + // PREPARE/DISCARD/DISCARDALL have no key and are not slot-routable. + $command = $commands->create('HIMPORT', ['PREPARE', 'fieldset', 'f1', 'f2']); + $this->assertNull($strategy->getSlot($command)); + + $command = $commands->create('HIMPORT', ['DISCARD', 'fieldset']); + $this->assertNull($strategy->getSlot($command)); + + $command = $commands->create('HIMPORT', ['DISCARDALL']); + $this->assertNull($strategy->getSlot($command)); + } + /** * @group disconnected */ @@ -716,6 +739,7 @@ class PredisStrategyTest extends PredisTestCase 'HVALS' => 'keys-first', 'HSCAN' => 'keys-first', 'HSTRLEN' => 'keys-first', + 'HIMPORT' => 'keys-himport', 'HEXPIRE' => 'keys-hash-field', 'HEXPIREAT' => 'keys-hash-field', 'HPERSIST' => 'keys-hash-field', diff --git a/tests/Predis/Cluster/RedisStrategyTest.php b/tests/Predis/Cluster/RedisStrategyTest.php index 940d2bf7..fd12ec6b 100644 --- a/tests/Predis/Cluster/RedisStrategyTest.php +++ b/tests/Predis/Cluster/RedisStrategyTest.php @@ -360,6 +360,29 @@ class RedisStrategyTest extends PredisTestCase } } + /** + * @group disconnected + */ + public function testKeysForHimportCommand(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // HIMPORT SET is routed by the hash slot of its key (at position 1). + $command = $commands->create('HIMPORT', ['SET', '{key}:1', 'fieldset', 'v1', 'v2']); + $this->assertNotNull($strategy->getSlot($command)); + + // PREPARE/DISCARD/DISCARDALL have no key and are not slot-routable. + $command = $commands->create('HIMPORT', ['PREPARE', 'fieldset', 'f1', 'f2']); + $this->assertNull($strategy->getSlot($command)); + + $command = $commands->create('HIMPORT', ['DISCARD', 'fieldset']); + $this->assertNull($strategy->getSlot($command)); + + $command = $commands->create('HIMPORT', ['DISCARDALL']); + $this->assertNull($strategy->getSlot($command)); + } + /** * @group disconnected */ @@ -736,6 +759,7 @@ class RedisStrategyTest extends PredisTestCase 'HVALS' => 'keys-first', 'HSCAN' => 'keys-first', 'HSTRLEN' => 'keys-first', + 'HIMPORT' => 'keys-himport', 'HEXPIRE' => 'keys-hash-field', 'HEXPIREAT' => 'keys-hash-field', 'HPERSIST' => 'keys-hash-field', diff --git a/tests/Predis/Command/Container/HIMPORT_Test.php b/tests/Predis/Command/Container/HIMPORT_Test.php new file mode 100644 index 00000000..584edc65 --- /dev/null +++ b/tests/Predis/Command/Container/HIMPORT_Test.php @@ -0,0 +1,530 @@ +options = new Options(); + $this->client = $this->getMockBuilder(ClientInterface::class)->getMock(); + $this->client->method('getOptions')->willReturn($this->options); + $this->client + ->method('createCommand') + ->willReturnCallback(function ($commandID, $arguments = []) { + return $this->createCommandStub($arguments); + }); + } + + /** + * @group disconnected + */ + public function testGetContainerCommandId(): void + { + $this->assertSame('HIMPORT', $this->createContainer()->getContainerCommandId()); + } + + /** + * @group disconnected + */ + public function testPrepareExecutesAndRecordsRegistry(): void + { + $node = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $this->client->method('getConnection')->willReturn($node); + $this->client->expects($this->once())->method('executeCommand')->willReturn(Status::get('OK')); + + $container = $this->createContainer(); + $response = $container->prepare('shared', ['name', 'email']); + + $this->assertEquals(Status::get('OK'), $response); + $this->assertSame(['name', 'email'], $this->options->himport->getRegistry()->get('shared')); + } + + /** + * @group disconnected + */ + public function testPrepareRejectsEmptyFields(): void + { + $this->expectException(InvalidArgumentException::class); + + $this->createContainer()->prepare('shared', []); + } + + /** + * @group disconnected + */ + public function testSetRejectsEmptyValues(): void + { + $this->expectException(InvalidArgumentException::class); + + $this->createContainer()->set('key', 'shared', []); + } + + /** + * @group disconnected + */ + public function testSetRecoversFromNoSuchFieldsetAndRetriesOnce(): void + { + $this->options->himport->getRegistry()->set('shared', ['name', 'email']); + + $node = $this->nodeWithRetries(1); + $node->expects($this->once())->method('executeCommand')->willReturn(Status::get('OK')); + $this->client->method('getConnection')->willReturn($node); + + $attempts = 0; + $this->client + ->expects($this->exactly(2)) + ->method('executeCommand') + ->willReturnCallback(function () use (&$attempts) { + if (0 === $attempts++) { + throw new ServerException('ERR no such fieldset'); + } + + return Status::get('OK'); + }); + + $response = $this->createContainer()->set('shared:1', 'shared', ['alice', 'alice@example.com']); + + $this->assertEquals(Status::get('OK'), $response); + } + + /** + * @group disconnected + */ + public function testSetDoesNotRetryWhenRetriesAreDisabled(): void + { + $this->options->himport->getRegistry()->set('shared', ['name']); + + // Retries disabled (the default): no re-prepare, no retry. + $node = $this->nodeWithRetries(0); + $node->expects($this->never())->method('executeCommand'); + $this->client->method('getConnection')->willReturn($node); + + $this->client + ->expects($this->once()) + ->method('executeCommand') + ->willThrowException(new ServerException('ERR no such fieldset')); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + + $this->createContainer()->set('shared:1', 'shared', ['alice']); + } + + /** + * @group disconnected + */ + public function testSetRecoveryIsBoundedByTheConfiguredRetryCount(): void + { + $this->options->himport->getRegistry()->set('shared', ['name']); + + $node = $this->nodeWithRetries(1); + $node->method('executeCommand')->willReturn(Status::get('OK')); + $this->client->method('getConnection')->willReturn($node); + + // One configured retry => the SET is attempted exactly twice. + $this->client + ->expects($this->exactly(2)) + ->method('executeCommand') + ->willThrowException(new ServerException('ERR no such fieldset')); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + + $this->createContainer()->set('shared:1', 'shared', ['alice']); + } + + /** + * @group disconnected + */ + public function testSetDoesNotRecoverWithoutRegistryEntry(): void + { + $node = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $this->client->method('getConnection')->willReturn($node); + + $this->client + ->expects($this->once()) + ->method('executeCommand') + ->willThrowException(new ServerException('ERR no such fieldset')); + + $this->expectException(ServerException::class); + + $this->createContainer()->set('shared:1', 'shared', ['alice']); + } + + /** + * @group disconnected + */ + public function testSetDoesNotRecoverWhenAutoPrepareDisabled(): void + { + $this->options = new Options(['himport' => ['auto_prepare' => false]]); + $this->client = $this->getMockBuilder(ClientInterface::class)->getMock(); + $this->client->method('getOptions')->willReturn($this->options); + $this->client + ->method('createCommand') + ->willReturnCallback(function ($commandID, $arguments = []) { + return $this->createCommandStub($arguments); + }); + $this->options->himport->getRegistry()->set('shared', ['name']); + + $node = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $this->client->method('getConnection')->willReturn($node); + $this->client + ->expects($this->once()) + ->method('executeCommand') + ->willThrowException(new ServerException('ERR no such fieldset')); + + $this->expectException(ServerException::class); + + $this->createContainer()->set('shared:1', 'shared', ['alice']); + } + + /** + * @group disconnected + */ + public function testSetRecoversFromErrorResponseWhenExceptionsDisabled(): void + { + $this->options = new Options(['exceptions' => false]); + $this->client = $this->getMockBuilder(ClientInterface::class)->getMock(); + $this->client->method('getOptions')->willReturn($this->options); + $this->client + ->method('createCommand') + ->willReturnCallback(function ($commandID, $arguments = []) { + return $this->createCommandStub($arguments); + }); + $this->options->himport->getRegistry()->set('shared', ['name']); + + $node = $this->nodeWithRetries(1); + $node->method('executeCommand')->willReturn(Status::get('OK')); + $this->client->method('getConnection')->willReturn($node); + + $attempts = 0; + $this->client + ->expects($this->exactly(2)) + ->method('executeCommand') + ->willReturnCallback(function () use (&$attempts) { + return 0 === $attempts++ ? new Error('ERR no such fieldset') : Status::get('OK'); + }); + + $response = $this->createContainer()->set('shared:1', 'shared', ['alice']); + + $this->assertEquals(Status::get('OK'), $response); + } + + /** + * A re-prepare that the server rejects must honour the exceptions option: + * with exceptions off it returns the Error response (root cause), not throws. + * + * @group disconnected + */ + public function testSetReturnsErrorWhenReprepareFailsAndExceptionsDisabled(): void + { + $this->options = new Options(['exceptions' => false]); + $this->client = $this->getMockBuilder(ClientInterface::class)->getMock(); + $this->client->method('getOptions')->willReturn($this->options); + $this->client + ->method('createCommand') + ->willReturnCallback(function ($commandID, $arguments = []) { + return $this->createCommandStub($arguments); + }); + $this->options->himport->getRegistry()->set('shared', ['name']); + + // The re-PREPARE (executed directly on the node) is rejected by the server. + $node = $this->nodeWithRetries(1); + $node->method('executeCommand')->willReturn(new Error('ERR duplicate field name in fieldset')); + $this->client->method('getConnection')->willReturn($node); + + // The SET itself reports "no such fieldset", triggering recovery. + $this->client + ->expects($this->once()) + ->method('executeCommand') + ->willReturn(new Error('ERR no such fieldset')); + + $response = $this->createContainer()->set('shared:1', 'shared', ['alice']); + + $this->assertInstanceOf(Error::class, $response); + $this->assertStringContainsString('duplicate field name', $response->getMessage()); + } + + /** + * The mirror case: with exceptions on, a rejected re-prepare surfaces the + * root cause as a thrown ServerException. + * + * @group disconnected + */ + public function testSetThrowsWhenReprepareFailsAndExceptionsEnabled(): void + { + $this->options->himport->getRegistry()->set('shared', ['name']); + + $node = $this->nodeWithRetries(1); + $node->method('executeCommand')->willReturn(new Error('ERR duplicate field name in fieldset')); + $this->client->method('getConnection')->willReturn($node); + + $this->client + ->expects($this->once()) + ->method('executeCommand') + ->willThrowException(new ServerException('ERR no such fieldset')); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage('duplicate field name'); + + $this->createContainer()->set('shared:1', 'shared', ['alice']); + } + + /** + * @group disconnected + */ + public function testPrepareFansOutToAllMasters(): void + { + [$node1, $node2] = $this->twoNodes(); + $node1->expects($this->once())->method('executeCommand')->willReturn(Status::get('OK')); + $node2->expects($this->once())->method('executeCommand')->willReturn(Status::get('OK')); + + $this->client->method('getConnection')->willReturn($this->clusterOf($node1, $node2)); + + $response = $this->createContainer()->prepare('shared', ['name']); + + $this->assertEquals(Status::get('OK'), $response); + $this->assertTrue($this->options->himport->getRegistry()->has('shared')); + } + + /** + * @group disconnected + */ + public function testPrepareFanOutSurfacesFirstErrorAndDoesNotRecord(): void + { + [$node1, $node2] = $this->twoNodes(); + $node1->method('executeCommand')->willReturn(Status::get('OK')); + $node2->method('executeCommand')->willReturn(new Error('ERR duplicate field name in fieldset')); + + $this->client->method('getConnection')->willReturn($this->clusterOf($node1, $node2)); + + try { + $this->createContainer()->prepare('shared', ['name', 'name']); + $this->fail('Expected ServerException was not thrown'); + } catch (ServerException $exception) { + $this->assertStringContainsString('duplicate field name', $exception->getMessage()); + } + + // A rejected PREPARE must not leave a poison registry entry that later + // SETs would keep trying (and failing) to re-prepare. + $this->assertFalse($this->options->himport->getRegistry()->has('shared')); + } + + /** + * @group disconnected + */ + public function testPrepareDoesNotRecordWhenStandaloneServerRejectsIt(): void + { + $node = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $this->client->method('getConnection')->willReturn($node); + $this->client + ->expects($this->once()) + ->method('executeCommand') + ->willThrowException(new ServerException('ERR duplicate field name in fieldset')); + + try { + $this->createContainer()->prepare('shared', ['name', 'name']); + $this->fail('Expected ServerException was not thrown'); + } catch (ServerException $exception) { + $this->assertStringContainsString('duplicate field name', $exception->getMessage()); + } + + $this->assertFalse($this->options->himport->getRegistry()->has('shared')); + } + + /** + * @group disconnected + */ + public function testDiscardKeepsRegistryWhenServerRejectsIt(): void + { + $this->options->himport->getRegistry()->set('shared', ['name']); + + $node = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $this->client->method('getConnection')->willReturn($node); + $this->client + ->method('executeCommand') + ->willThrowException(new ServerException('ERR something went wrong')); + + try { + $this->createContainer()->discard('shared'); + $this->fail('Expected ServerException was not thrown'); + } catch (ServerException $exception) { + // A failed discard leaves the fieldset consistently known and retryable. + } + + $this->assertTrue($this->options->himport->getRegistry()->has('shared')); + } + + /** + * A ClusterInterface that cannot be iterated must fail fast rather than look + * like a successful fan-out over zero nodes. + * + * @group disconnected + */ + public function testPrepareFailsFastWhenClusterIsNotIterable(): void + { + // A bare ClusterInterface mock does not implement IteratorAggregate. + $cluster = $this->getMockBuilder(ClusterInterface::class)->getMock(); + $this->client->method('getConnection')->willReturn($cluster); + + try { + $this->createContainer()->prepare('shared', ['name']); + $this->fail('Expected a NotSupportedException was not thrown'); + } catch (NotSupportedException $exception) { + $this->assertStringContainsString('iterable', $exception->getMessage()); + } + + // Nothing was executed, so nothing must have been recorded. + $this->assertFalse($this->options->himport->getRegistry()->has('shared')); + } + + /** + * An iterable cluster that yields no master shards must also fail fast. + * + * @group disconnected + */ + public function testPrepareFailsFastWhenClusterHasNoMasters(): void + { + $cluster = $this->getMockBuilder(RedisCluster::class) + ->disableOriginalConstructor() + ->onlyMethods(['getIterator']) + ->getMock(); + $cluster->method('getIterator')->willReturn(new ArrayIterator([])); + $this->client->method('getConnection')->willReturn($cluster); + + $this->expectException(NotSupportedException::class); + + $this->createContainer()->prepare('shared', ['name']); + } + + /** + * @group disconnected + */ + public function testDiscardFanOutReturnsMaxAndClearsRegistry(): void + { + $this->options->himport->getRegistry()->set('shared', ['name']); + + [$node1, $node2] = $this->twoNodes(); + $node1->method('executeCommand')->willReturn(1); + $node2->method('executeCommand')->willReturn(0); + + $this->client->method('getConnection')->willReturn($this->clusterOf($node1, $node2)); + + $this->assertSame(1, $this->createContainer()->discard('shared')); + $this->assertFalse($this->options->himport->getRegistry()->has('shared')); + } + + /** + * @group disconnected + */ + public function testDiscardAllFanOutReturnsMaxAndClearsRegistry(): void + { + $this->options->himport->getRegistry()->set('fs1', ['a']); + $this->options->himport->getRegistry()->set('fs2', ['b']); + + [$node1, $node2] = $this->twoNodes(); + $node1->method('executeCommand')->willReturn(2); + $node2->method('executeCommand')->willReturn(2); + + $this->client->method('getConnection')->willReturn($this->clusterOf($node1, $node2)); + + $this->assertSame(2, $this->createContainer()->discardAll()); + $this->assertEmpty($this->options->himport->getRegistry()->all()); + } + + private function createContainer(): HIMPORT + { + return new HIMPORT($this->client); + } + + /** + * A node connection whose parameters carry a Retry configured with the given + * number of retries (0 = disabled), so the container reuses it as-is. + * + * @param int $retries + * @return MockObject&NodeConnectionInterface + */ + private function nodeWithRetries(int $retries) + { + $node = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $node + ->method('getParameters') + ->willReturn(new Parameters(['retry' => new Retry(new NoBackoff(), $retries)])); + + return $node; + } + + /** + * @return array{0: MockObject&NodeConnectionInterface, 1: MockObject&NodeConnectionInterface} + */ + private function twoNodes(): array + { + return [ + $this->getMockBuilder(NodeConnectionInterface::class)->getMock(), + $this->getMockBuilder(NodeConnectionInterface::class)->getMock(), + ]; + } + + private function clusterOf(NodeConnectionInterface $node1, NodeConnectionInterface $node2): RedisCluster + { + $cluster = $this->getMockBuilder(RedisCluster::class) + ->disableOriginalConstructor() + ->onlyMethods(['getIterator']) + ->getMock(); + $cluster->method('getIterator')->willReturn(new ArrayIterator([$node1, $node2])); + + return $cluster; + } + + /** + * @param array $arguments + * @return CommandInterface + */ + private function createCommandStub(array $arguments): CommandInterface + { + $command = $this->getMockBuilder(CommandInterface::class)->getMock(); + $command->method('getId')->willReturn('HIMPORT'); + $command->method('getArguments')->willReturn($arguments); + + return $command; + } +} diff --git a/tests/Predis/Command/Redis/HIMPORT_Test.php b/tests/Predis/Command/Redis/HIMPORT_Test.php new file mode 100644 index 00000000..92a4db23 --- /dev/null +++ b/tests/Predis/Command/Redis/HIMPORT_Test.php @@ -0,0 +1,910 @@ +getCommand(); + $command->setArguments($actualArguments); + + $this->assertSame($expectedArguments, $command->getArguments()); + } + + /** + * @group disconnected + */ + public function testParseResponse(): void + { + $this->assertSame(1, $this->getCommand()->parseResponse(1)); + $this->assertEquals(new Status('OK'), $this->getCommand()->parseResponse(new Status('OK'))); + } + + /** + * @group disconnected + * @dataProvider prefixKeysProvider + */ + public function testPrefixKeys(array $arguments, array $expectedArguments): void + { + $command = $this->getCommand(); + $command->setArguments($arguments); + $command->prefixKeys('prefix:'); + + $this->assertSame($expectedArguments, $command->getArguments()); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testPrepareAndSetCreateHashFromSharedFieldset(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->himport->prepare('shared', ['name', 'email', 'age'])); + $this->assertEquals('OK', $redis->himport->set('shared:1', 'shared', ['alice', 'alice@example.com', '25'])); + + $this->assertSame('alice', $redis->hget('shared:1', 'name')); + + // Hash enumeration order is not guaranteed to match the PREPARE order + // (the server keeps a canonical internal order), so compare sorted. + $hash = $redis->hgetall('shared:1'); + ksort($hash); + $this->assertSame(['age' => '25', 'email' => 'alice@example.com', 'name' => 'alice'], $hash); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testValuesMapByEachFieldsetsOwnOrder(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('order1', ['a', 'b', 'c']); + $redis->himport->prepare('order2', ['c', 'b', 'a']); + + $redis->himport->set('order:key1', 'order1', ['va1', 'vb1', 'vc1']); + $redis->himport->set('order:key2', 'order2', ['vc2', 'vb2', 'va2']); + + $this->assertSame('va1', $redis->hget('order:key1', 'a')); + $this->assertSame('va2', $redis->hget('order:key2', 'a')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testSetFullyReplacesExistingHash(): void + { + $redis = $this->getClient(); + + $redis->hset('replace:1', 'stale', 'value'); + $redis->himport->prepare('shared', ['name', 'age']); + $redis->himport->set('replace:1', 'shared', ['bob', '30']); + + // Full replace: the pre-existing "stale" field must be gone. + $hash = $redis->hgetall('replace:1'); + ksort($hash); + $this->assertSame(['age' => '30', 'name' => 'bob'], $hash); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testDiscardReturnsRemovalCount(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('shared', ['name']); + + $this->assertSame(1, $redis->himport->discard('shared')); + $this->assertSame(0, $redis->himport->discard('shared')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testDiscardAllReturnsRemovalCount(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('fs1', ['a']); + $redis->himport->prepare('fs2', ['b']); + + $this->assertSame(2, $redis->himport->discardAll()); + $this->assertSame(0, $redis->himport->discardAll()); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testPrepareAndSetResp3(): void + { + $redis = $this->getResp3Client(); + + $this->assertEquals('OK', $redis->himport->prepare('shared', ['name', 'age'])); + $this->assertEquals('OK', $redis->himport->set('shared:1', 'shared', ['carol', '40'])); + + $this->assertSame('carol', $redis->hget('shared:1', 'name')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testThrowsExceptionOnDuplicateFieldName(): void + { + $this->expectException(ServerException::class); + $this->expectExceptionMessage('duplicate field name'); + + $redis = $this->getClient(); + $redis->himport->prepare('shared', ['name', 'name']); + } + + /** + * A PREPARE the server rejects must not leave a registry entry: a later SET + * reports the fieldset as unknown rather than re-raising the prepare error. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testRejectedPrepareDoesNotPoisonRegistry(): void + { + $redis = $this->getClient(); + + try { + $redis->himport->prepare('bad', ['name', 'name']); + $this->fail('Expected a "duplicate field name" error'); + } catch (ServerException $exception) { + $this->assertStringContainsString('duplicate field name', $exception->getMessage()); + } + + try { + $redis->himport->set('bad:1', 'bad', ['alice']); + $this->fail('Expected a "no such fieldset" error'); + } catch (ServerException $exception) { + $this->assertStringContainsString('no such fieldset', $exception->getMessage()); + $this->assertStringNotContainsString('duplicate field name', $exception->getMessage()); + } + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testThrowsExceptionOnUnknownFieldset(): void + { + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + + // Fieldset never declared through the container: nothing to recover + // from, so the server error is propagated unchanged. + $redis = $this->getClient(); + $redis->himport->set('shared:1', 'missing', ['value']); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testThrowsExceptionOnWrongType(): void + { + $this->expectException(ServerException::class); + $this->expectExceptionMessage('Operation against a key holding the wrong kind of value'); + + $redis = $this->getClient(); + + $redis->set('himport:string', 'a'); + $redis->himport->prepare('shared', ['name']); + $redis->himport->set('himport:string', 'shared', ['value']); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testRawCommandFormIsSupported(): void + { + $redis = $this->getClient(); + + // The lower-level raw form (no client-side registry or auto-recovery) + // remains available for pipelines and explicit connection control. + $this->assertEquals('OK', $redis->himport('PREPARE', 'shared', 'name', 'age')); + $this->assertEquals('OK', $redis->himport('SET', 'shared:1', 'shared', 'dave', '50')); + $this->assertSame('dave', $redis->hget('shared:1', 'name')); + $this->assertSame(1, $redis->himport('DISCARD', 'shared')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testMultiplePreparesAndDiscardsKeepRegistryAndSessionCommandsCoherent(): void + { + $redis = $this->getClient(); + $registry = $redis->getOptions()->himport->getRegistry(); + $connection = $redis->getConnection(); + + $redis->himport->prepare('fs1', ['a']); + $redis->himport->prepare('fs2', ['b', 'c']); + $redis->himport->prepare('fs3', ['d']); + + // Both the client-side registry and the connection's replay queue track + // every live fieldset, keyed and insertion-ordered. + $this->assertSame(['fs1', 'fs2', 'fs3'], array_keys($registry->all())); + $this->assertSame( + ['himport:fs1', 'himport:fs2', 'himport:fs3'], + array_keys($connection->getSessionCommands()) + ); + + // A single discard removes exactly one fieldset from both stores. + $this->assertSame(1, $redis->himport->discard('fs2')); + $this->assertSame(['fs1', 'fs3'], array_keys($registry->all())); + $this->assertSame( + ['himport:fs1', 'himport:fs3'], + array_keys($connection->getSessionCommands()) + ); + + // Re-preparing an existing name replaces its field list (last wins). + $redis->himport->prepare('fs1', ['a', 'x']); + $this->assertSame(['a', 'x'], $registry->get('fs1')); + + // DISCARDALL clears both stores and reports the removal count. + $this->assertSame(2, $redis->himport->discardAll()); + $this->assertSame([], $registry->all()); + $this->assertSame([], $connection->getSessionCommands()); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testReconnectTransparentlyReplaysPreparedFieldsets(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('shared', ['name', 'age']); + + // Simulate a dropped connection mid-ingestion. The pinned PREPARE is + // replayed when the connection is re-established, so the SET succeeds. + $redis->getConnection()->disconnect(); + + $this->assertEquals('OK', $this->setAfterReconnect($redis, 'shared:1', 'shared', ['erin', '28'])); + $this->assertSame('erin', $redis->hget('shared:1', 'name')); + $this->assertSame('28', $redis->hget('shared:1', 'age')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testReactiveRecoveryRePreparesWhenReplayIsUnavailable(): void + { + // Recovery reuses the connection's Retry policy, so it must be enabled. + $redis = $this->createClient(['retry' => new Retry(new NoBackoff(), 1)]); + + $redis->himport->prepare('shared', ['name', 'age']); + + // Drop the proactive replay entry to emulate a connection that never + // received the PREPARE (e.g. a new cluster node reached via redirection). + $redis->getConnection()->removeSessionCommandsByPrefix('himport:'); + $redis->getConnection()->disconnect(); + + // The first SET hits "no such fieldset"; the container re-prepares from + // the registry on the executing connection and retries exactly once. + $this->assertEquals('OK', $this->setAfterReconnect($redis, 'shared:1', 'shared', ['frank', '33'])); + $this->assertSame('frank', $redis->hget('shared:1', 'name')); + $this->assertSame('33', $redis->hget('shared:1', 'age')); + + // Recovery re-pins the command for subsequent reconnects. + $this->assertArrayHasKey('himport:shared', $redis->getConnection()->getSessionCommands()); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testAutoPrepareDisabledPropagatesErrorWithoutRecovering(): void + { + $redis = $this->createClient(null, ['himport' => ['auto_prepare' => false]]); + + $redis->himport->prepare('shared', ['name', 'age']); + $redis->getConnection()->removeSessionCommandsByPrefix('himport:'); + $redis->getConnection()->disconnect(); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + + $this->setAfterReconnect($redis, 'shared:1', 'shared', ['grace', '41']); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testPipelineRunsPrepareAndSetsOnASingleConnection(): void + { + $redis = $this->getClient(); + + // The intended high-throughput pattern: PREPARE followed by dependent + // SETs, back-to-back on one connection, using the raw command form. + $responses = $redis->pipeline(function ($pipe) { + $pipe->himport('PREPARE', 'shared', 'name', 'age'); + $pipe->himport('SET', 'shared:1', 'shared', 'heidi', '22'); + $pipe->himport('SET', 'shared:2', 'shared', 'ivan', '44'); + }); + + $this->assertEquals('OK', $responses[0]); + $this->assertEquals('OK', $responses[1]); + $this->assertEquals('OK', $responses[2]); + $this->assertSame('heidi', $redis->hget('shared:1', 'name')); + $this->assertSame('ivan', $redis->hget('shared:2', 'name')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testTransactionRunsPrepareAndSetOnASingleConnection(): void + { + $redis = $this->getClient(); + + $responses = $redis->transaction(function ($tx) { + $tx->himport('PREPARE', 'shared', 'name', 'age'); + $tx->himport('SET', 'shared:1', 'shared', 'judy', '37'); + }); + + $this->assertEquals('OK', $responses[0]); + $this->assertEquals('OK', $responses[1]); + $hash = $redis->hgetall('shared:1'); + ksort($hash); + $this->assertSame(['age' => '37', 'name' => 'judy'], $hash); + } + + /** + * Full lifecycle: PREPARE enables SET, DISCARD ends it, and a SET issued + * afterwards raises the server error (the registry is cleared by DISCARD, so + * auto-recovery does not mask it). Keys already written survive the discard. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testFieldsetLifecycleFromPrepareThroughDiscard(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('shared', ['name', 'age']); + $this->assertEquals('OK', $redis->himport->set('shared:1', 'shared', ['alice', '25'])); + $this->assertEquals('OK', $redis->himport->set('shared:2', 'shared', ['bob', '30'])); + $this->assertSame('alice', $redis->hget('shared:1', 'name')); + + $this->assertSame(1, $redis->himport->discard('shared')); + + // Keys created through the fieldset are ordinary hashes and are not + // affected by discarding it. + $this->assertSame('bob', $redis->hget('shared:2', 'name')); + + // From here on, SET against the discarded fieldset must error. + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + $redis->himport->set('shared:3', 'shared', ['carol', '40']); + } + + /** + * DISCARDALL ends every fieldset at once; each subsequent SET errors. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testDiscardAllStopsEverySubsequentSet(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('fs1', ['a']); + $redis->himport->prepare('fs2', ['b']); + $this->assertEquals('OK', $redis->himport->set('k1', 'fs1', ['1'])); + $this->assertEquals('OK', $redis->himport->set('k2', 'fs2', ['2'])); + + $this->assertSame(2, $redis->himport->discardAll()); + + // Previously written keys survive. + $this->assertSame('1', $redis->hget('k1', 'a')); + + foreach (['fs1', 'fs2'] as $fieldset) { + try { + $redis->himport->set('k_' . $fieldset, $fieldset, ['9']); + $this->fail("Expected 'no such fieldset' for {$fieldset}"); + } catch (ServerException $exception) { + $this->assertStringContainsString('no such fieldset', $exception->getMessage()); + } + } + } + + /** + * A fieldset already present on the connection but never declared through the + * container (prepared out of band with the raw command, or inherited on a + * persistent connection) is usable by the container's set(): the write only + * needs the fieldset on the executing connection, not a registry entry. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testContainerSetUsesFieldsetPreparedOutOfBand(): void + { + $redis = $this->getClient(); + + // Prepared with the raw form: the server-side fieldset exists on the + // connection, but the client-side registry is left untouched. + $this->assertEquals('OK', $redis->himport('PREPARE', 'shared', 'name', 'age')); + $this->assertFalse($redis->getOptions()->himport->getRegistry()->has('shared')); + + $this->assertEquals('OK', $redis->himport->set('shared:1', 'shared', ['alice', '25'])); + $this->assertSame('alice', $redis->hget('shared:1', 'name')); + + // Still not tracked: the container used the pre-loaded state as-is. + $this->assertFalse($redis->getOptions()->himport->getRegistry()->has('shared')); + } + + /** + * Recovery is scoped to fieldsets declared through the container. A pre-loaded + * fieldset the registry does not know about is NOT re-prepared after its + * connection is lost, so the server error surfaces unchanged. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testPreloadedFieldsetIsNotRecoveredAfterReconnect(): void + { + $redis = $this->getClient(); + + $redis->himport('PREPARE', 'shared', 'name', 'age'); + $this->assertEquals('OK', $redis->himport->set('shared:1', 'shared', ['alice', '25'])); + + // The raw PREPARE pinned nothing for replay and recorded nothing in the + // registry, so after a reconnect the fieldset is simply gone. + $redis->getConnection()->disconnect(); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + $this->setAfterReconnect($redis, 'shared:2', 'shared', ['bob', '30']); + } + + /** + * A fieldset declared through the `himport` client option is prepared on + * demand: a SET referencing it succeeds without an explicit prepare() call. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testConfiguredFieldsetIsPreparedOnDemandWithoutExplicitPrepare(): void + { + // On-demand preparation reuses the connection's Retry policy. + $redis = $this->createClient(['retry' => new Retry(new NoBackoff(), 1)], [ + 'himport' => ['fieldsets' => ['users' => ['name', 'age']]], + ]); + + // No prepare() call - the fieldset is known from configuration. + $this->assertEquals('OK', $redis->himport->set('users:1', 'users', ['alice', '25'])); + $this->assertEquals('OK', $redis->himport->set('users:2', 'users', ['bob', '30'])); + + $this->assertSame('alice', $redis->hget('users:1', 'name')); + $this->assertSame('bob', $redis->hget('users:2', 'name')); + } + + /** + * On-demand preparation of configured fieldsets uses the auto-prepare + * mechanism; with it disabled and no explicit prepare(), the server error + * is propagated. + * + * @group connected + * @requiresRedisVersion >= 8.9.0 + */ + public function testConfiguredFieldsetWithAutoPrepareDisabledIsNotPrepared(): void + { + $redis = $this->createClient(null, [ + 'himport' => ['fieldsets' => ['users' => ['name', 'age']], 'auto_prepare' => false], + ]); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage('no such fieldset'); + $redis->himport->set('users:1', 'users', ['alice', '25']); + } + + /** + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterFansOutPrepareToEveryMasterAndPinsPerNode(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->himport->prepare('shared', ['name', 'email', 'age'])); + + // PREPARE is executed on every master shard and pinned on each node so + // it survives that node's future reconnects. + $masters = iterator_to_array($redis->getConnection()->getIterator()); + $this->assertGreaterThan(1, count($masters)); + + foreach ($masters as $master) { + $this->assertArrayHasKey('himport:shared', $master->getSessionCommands()); + } + } + + /** + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterRoutesSetByKeySlotAcrossShards(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('shared', ['name', 'age']); + + // Keys chosen to land on different shards; each master already holds the + // fanned-out fieldset, so every SET succeeds by normal slot routing. + foreach (['shared:{a}', 'shared:{b}', 'shared:{c}', 'shared:{d}'] as $index => $key) { + $this->assertEquals('OK', $redis->himport->set($key, 'shared', ['user' . $index, (string) (20 + $index)])); + $this->assertSame('user' . $index, $redis->hget($key, 'name')); + } + } + + /** + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterMultiplePreparesAndDiscardManageStatePerNode(): void + { + $redis = $this->getClient(); + + $redis->himport->prepare('fs1', ['a']); + $redis->himport->prepare('fs2', ['b']); + + $registry = $redis->getOptions()->himport->getRegistry(); + $this->assertSame(['fs1', 'fs2'], array_keys($registry->all())); + + foreach (iterator_to_array($redis->getConnection()->getIterator()) as $master) { + $this->assertSame( + ['himport:fs1', 'himport:fs2'], + array_keys($master->getSessionCommands()) + ); + } + + // Discarding fs1 fans out to every master and updates every store. + $this->assertSame(1, $redis->himport->discard('fs1')); + $this->assertSame(['fs2'], array_keys($registry->all())); + + foreach (iterator_to_array($redis->getConnection()->getIterator()) as $master) { + $this->assertSame(['himport:fs2'], array_keys($master->getSessionCommands())); + } + + $this->assertSame(1, $redis->himport->discardAll()); + $this->assertSame([], $registry->all()); + + foreach (iterator_to_array($redis->getConnection()->getIterator()) as $master) { + $this->assertSame([], $master->getSessionCommands()); + } + } + + /** + * Full cluster lifecycle across shards: after fan-out PREPARE, SET succeeds + * for keys on every shard; after fan-out DISCARD, SET errors on every shard, + * while hashes already written survive. + * + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterFieldsetLifecycleAcrossShards(): void + { + $redis = $this->getClient(); + $redis->himport->prepare('shared', ['name', 'age']); + + $keys = ['shared:{a}', 'shared:{b}', 'shared:{c}', 'shared:{d}']; + + foreach ($keys as $i => $key) { + $this->assertEquals('OK', $redis->himport->set($key, 'shared', ['user' . $i, (string) (20 + $i)])); + } + foreach ($keys as $i => $key) { + $this->assertSame('user' . $i, $redis->hget($key, 'name')); + } + + // DISCARD fans out to every master; the fieldset is gone cluster-wide. + $this->assertSame(1, $redis->himport->discard('shared')); + + // Keys written earlier are unaffected. + $this->assertSame('user0', $redis->hget($keys[0], 'name')); + + // A SET on any shard now errors and is not auto-recovered (registry cleared). + foreach ($keys as $key) { + try { + $redis->himport->set($key, 'shared', ['x', '1']); + $this->fail("Expected 'no such fieldset' for {$key}"); + } catch (ServerException $exception) { + $this->assertStringContainsString('no such fieldset', $exception->getMessage()); + } + } + } + + /** + * Per-node state independence: with auto-prepare OFF, removing the fieldset + * from ONE master makes SETs routed to that master fail, while SETs routed to + * other masters keep working. Proves the error appears exactly where the + * server-side state is missing and nowhere else. + * + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterPerNodeStateWithoutAutoPrepareErrorsOnlyWhereMissing(): void + { + $redis = $this->createClient(null, ['himport' => ['auto_prepare' => false]]); + $redis->himport->prepare('shared', ['name']); + + $cluster = $redis->getConnection(); + $byNode = $this->groupKeysByNode($redis, $cluster); + + if (count($byNode) < 2) { + $this->markTestSkipped('Requires at least two distinct master shards.'); + } + + $ids = array_keys($byNode); + $victim = $byNode[$ids[0]]; + $other = $byNode[$ids[1]]; + + // Remove the fieldset from the victim master only (direct raw command). + $victim['node']->executeCommand(RawCommand::create('HIMPORT', 'DISCARD', 'shared')); + + // SET routed to the victim shard errors; no recovery because it is off. + try { + $redis->himport->set($victim['keys'][0], 'shared', ['alice']); + $this->fail('Expected a "no such fieldset" error on the victim shard'); + } catch (ServerException $exception) { + $this->assertStringContainsString('no such fieldset', $exception->getMessage()); + } + + // SET routed to a shard that still has the fieldset keeps working. + $this->assertEquals('OK', $redis->himport->set($other['keys'][0], 'shared', ['bob'])); + $this->assertSame('bob', $redis->hget($other['keys'][0], 'name')); + } + + /** + * With auto-prepare ON, a SET routed to a master that is missing the fieldset + * (e.g. a node recycled or reached via redirection) transparently re-prepares + * on that master and succeeds - no error surfaces where we do not want one. + * + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterAutoPrepareRecoversOnNodeMissingFieldset(): void + { + // Recovery reuses the connection's Retry policy, so enable it. + $redis = $this->createClient(null, ['parameters' => ['retry' => new Retry(new NoBackoff(), 1)]]); + $redis->himport->prepare('shared', ['name']); + + $cluster = $redis->getConnection(); + $key = 'shared:{a}'; + $node = $cluster->getConnectionByCommand( + $redis->createCommand('himport', ['SET', $key, 'shared', 'placeholder']) + ); + + // Wipe the fieldset from just the node that owns this key's slot. + $node->executeCommand(RawCommand::create('HIMPORT', 'DISCARD', 'shared')); + + // The SET hits "no such fieldset", re-prepares on that node, retries once. + $this->assertEquals('OK', $redis->himport->set($key, 'shared', ['carol'])); + $this->assertSame('carol', $redis->hget($key, 'name')); + } + + /** + * On a cluster, a fieldset pre-loaded directly on the master that owns a key + * (out of band, without the container fan-out or a registry entry) is usable + * by the container's set() for that key. + * + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterContainerSetUsesFieldsetPreloadedOnOwningNode(): void + { + $redis = $this->getClient(); + $cluster = $redis->getConnection(); + + $key = 'shared:{a}'; + $node = $cluster->getConnectionByCommand( + $redis->createCommand('himport', ['SET', $key, 'shared', 'placeholder']) + ); + + // Load the fieldset only on the node that owns this key's slot; the + // registry stays empty and no fan-out happens. + $node->executeCommand(RawCommand::create('HIMPORT', 'PREPARE', 'shared', 'name', 'age')); + $this->assertFalse($redis->getOptions()->himport->getRegistry()->has('shared')); + + $this->assertEquals('OK', $redis->himport->set($key, 'shared', ['dave', '44'])); + $this->assertSame('dave', $redis->hget($key, 'name')); + } + + /** + * A fieldset declared through the `himport` option is prepared on demand on + * whichever master owns each key - SETs across shards succeed with no + * explicit prepare() and no manual fan-out. + * + * @group connected + * @group cluster + * @requiresRedisVersion >= 8.9.0 + */ + public function testClusterConfiguredFieldsetPreparedOnDemandAcrossShards(): void + { + // On-demand preparation reuses the connection's Retry policy. + $redis = $this->createClient(null, [ + 'himport' => ['fieldsets' => ['users' => ['name', 'age']]], + 'parameters' => ['retry' => new Retry(new NoBackoff(), 1)], + ]); + + foreach (['users:{a}', 'users:{b}', 'users:{c}', 'users:{d}'] as $i => $key) { + $this->assertEquals('OK', $redis->himport->set($key, 'users', ['user' . $i, (string) (20 + $i)])); + } + foreach (['users:{a}', 'users:{b}', 'users:{c}', 'users:{d}'] as $i => $key) { + $this->assertSame('user' . $i, $redis->hget($key, 'name')); + } + } + + /** + * Executes an HIMPORT SET, reconnecting first when the backend does not + * reconnect on its own after an explicit disconnect(). The stream backend + * reconnects lazily on the next command, but ext-relay raises + * `RELAY_ERR_IO ("Not connected")` instead, so these tests catch that and + * reconnect to stay backend-agnostic. + * + * @param array $values + * @return mixed + */ + private function setAfterReconnect(ClientInterface $redis, string $key, string $fieldset, array $values) + { + try { + return $redis->himport->set($key, $fieldset, $values); + } catch (ConnectionException $exception) { + $redis->connect(); + + return $redis->himport->set($key, $fieldset, $values); + } + } + + /** + * Groups a spread of hash-tagged keys by the master shard that owns them. + * + * @param \Predis\Client $redis + * @param mixed $cluster Aggregate cluster connection. + * @return array + */ + private function groupKeysByNode($redis, $cluster): array + { + $byNode = []; + + foreach (range('a', 'p') as $tag) { + $key = 'shared:{' . $tag . '}'; + $node = $cluster->getConnectionByCommand( + $redis->createCommand('himport', ['SET', $key, 'shared', 'placeholder']) + ); + $id = (string) $node; + + if (!isset($byNode[$id])) { + $byNode[$id] = ['node' => $node, 'keys' => []]; + } + + $byNode[$id]['keys'][] = $key; + } + + return $byNode; + } + + public function argumentsProvider(): array + { + return [ + 'PREPARE with variadic fields' => [ + ['PREPARE', 'shared', 'name', 'email', 'age'], + ['PREPARE', 'shared', 'name', 'email', 'age'], + ], + 'PREPARE with fields as array' => [ + ['PREPARE', 'shared', ['name', 'email', 'age']], + ['PREPARE', 'shared', 'name', 'email', 'age'], + ], + 'SET with values as array' => [ + ['SET', 'shared:1', 'shared', ['alice', 'alice@example.com', '25']], + ['SET', 'shared:1', 'shared', 'alice', 'alice@example.com', '25'], + ], + 'field order preserved verbatim' => [ + ['PREPARE', 'order', ['c', 'a', 'b']], + ['PREPARE', 'order', 'c', 'a', 'b'], + ], + 'empty string fieldset and field names preserved' => [ + ['PREPARE', '', ['']], + ['PREPARE', '', ''], + ], + 'DISCARD' => [ + ['DISCARD', 'shared'], + ['DISCARD', 'shared'], + ], + 'DISCARDALL' => [ + ['DISCARDALL'], + ['DISCARDALL'], + ], + 'minimal placeholder shape tolerated' => [ + ['key'], + ['key'], + ], + ]; + } + + public function prefixKeysProvider(): array + { + return [ + 'SET prefixes the key at index 1 only' => [ + ['SET', 'shared:1', 'shared', 'alice', '25'], + ['SET', 'prefix:shared:1', 'shared', 'alice', '25'], + ], + 'PREPARE fieldset name is not prefixed' => [ + ['PREPARE', 'shared', 'name', 'age'], + ['PREPARE', 'shared', 'name', 'age'], + ], + 'DISCARD fieldset name is not prefixed' => [ + ['DISCARD', 'shared'], + ['DISCARD', 'shared'], + ], + 'DISCARDALL is untouched' => [ + ['DISCARDALL'], + ['DISCARDALL'], + ], + ]; + } +} diff --git a/tests/Predis/Configuration/Option/HimportTest.php b/tests/Predis/Configuration/Option/HimportTest.php new file mode 100644 index 00000000..7d8de1a1 --- /dev/null +++ b/tests/Predis/Configuration/Option/HimportTest.php @@ -0,0 +1,151 @@ +getMockBuilder(OptionsInterface::class)->getMock(); + } + + /** + * @group disconnected + */ + public function testDefaultEnablesAutoPrepareWithEmptyRegistry(): void + { + $option = new Himport(); + $default = $option->getDefault($this->getOptions()); + + $this->assertInstanceOf(HimportOptions::class, $default); + $this->assertTrue($default->isAutoPrepareEnabled()); + $this->assertSame([], $default->getRegistry()->all()); + } + + /** + * @group disconnected + */ + public function testFilterReturnsHimportOptionsInstanceAsIs(): void + { + $option = new Himport(); + $value = new HimportOptions(new FieldsetRegistry(), false); + + $this->assertSame($value, $option->filter($this->getOptions(), $value)); + } + + /** + * @group disconnected + */ + public function testFilterReadsAutoPrepareFlag(): void + { + $option = new Himport(); + + $this->assertFalse($option->filter($this->getOptions(), ['auto_prepare' => false])->isAutoPrepareEnabled()); + $this->assertTrue($option->filter($this->getOptions(), ['auto_prepare' => true])->isAutoPrepareEnabled()); + $this->assertTrue($option->filter($this->getOptions(), [])->isAutoPrepareEnabled()); + } + + /** + * @group disconnected + */ + public function testFilterSeedsRegistryFromPreloadedFieldsets(): void + { + $option = new Himport(); + + $result = $option->filter($this->getOptions(), [ + 'fieldsets' => [ + 'users' => ['name', 'email', 'age'], + 'orders' => ['id', 'total'], + ], + ]); + + $registry = $result->getRegistry(); + $this->assertTrue($registry->has('users')); + $this->assertTrue($registry->has('orders')); + $this->assertSame(['name', 'email', 'age'], $registry->get('users')); + $this->assertSame(['id', 'total'], $registry->get('orders')); + + // Pre-loading and the auto-prepare flag are independent inputs. + $this->assertTrue($result->isAutoPrepareEnabled()); + } + + /** + * @group disconnected + */ + public function testFilterAcceptsFieldsetsAndAutoPrepareTogether(): void + { + $option = new Himport(); + + $result = $option->filter($this->getOptions(), [ + 'fieldsets' => ['users' => ['name']], + 'auto_prepare' => false, + ]); + + $this->assertTrue($result->getRegistry()->has('users')); + $this->assertFalse($result->isAutoPrepareEnabled()); + } + + /** + * @group disconnected + */ + public function testFilterPreservesFieldOrderVerbatim(): void + { + $option = new Himport(); + + $result = $option->filter($this->getOptions(), [ + 'fieldsets' => ['ordered' => ['c', 'a', 'b']], + ]); + + $this->assertSame(['c', 'a', 'b'], $result->getRegistry()->get('ordered')); + } + + /** + * @group disconnected + */ + public function testFilterThrowsOnNonArrayValue(): void + { + $this->expectException(InvalidArgumentException::class); + + (new Himport())->filter($this->getOptions(), 'invalid'); + } + + /** + * @group disconnected + */ + public function testFilterThrowsWhenFieldsetsIsNotAMap(): void + { + $this->expectException(InvalidArgumentException::class); + + (new Himport())->filter($this->getOptions(), ['fieldsets' => 'invalid']); + } + + /** + * @group disconnected + */ + public function testFilterThrowsOnEmptyFieldList(): void + { + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage('Fieldset "users" must be a non-empty list of field names.'); + + (new Himport())->filter($this->getOptions(), ['fieldsets' => ['users' => []]]); + } +} diff --git a/tests/Predis/Connection/Replication/SentinelReplicationTest.php b/tests/Predis/Connection/Replication/SentinelReplicationTest.php index 0b04d0b5..4ccc2ffa 100644 --- a/tests/Predis/Connection/Replication/SentinelReplicationTest.php +++ b/tests/Predis/Connection/Replication/SentinelReplicationTest.php @@ -1564,6 +1564,77 @@ class SentinelReplicationTest extends PredisTestCase )); } + /** + * When every retry also fails with a StreamInitException, the original + * transport error must be surfaced unchanged - not masked by a TypeError in + * the retry fail callback (regression test for issue #1713). + * + * @group disconnected + */ + public function testMethodExecuteCommandSurfacesStreamInitExceptionWhenAllRetriesFail(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willReturn( + ['127.0.0.1', '6391'] + ); + + $masterOld = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $masterOld + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterOld + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('DEL', ['key'])) + ->willThrowException( + new StreamInitException('Connection refused [tcp://127.0.0.1:6381]') + ); + + $masterNew = $this->getMockConnection('tcp://127.0.0.1:6391?role=master'); + $masterNew + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterNew + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('DEL', ['key'])) + ->willThrowException( + new StreamInitException('Connection refused [tcp://127.0.0.1:6391]') + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6391', + 'role' => 'master', + ]) + ->willReturn($masterNew); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + $replication->add($masterOld); + $replication->setRetryLimit(1); + + // The fail callback is invoked with the StreamInitException on the first + // attempt (it must not TypeError), then the second attempt exhausts the + // retry budget and the original StreamInitException is re-thrown. + $this->expectException(StreamInitException::class); + $this->expectExceptionMessage('Connection refused [tcp://127.0.0.1:6391]'); + + $replication->executeCommand(Command\RawCommand::create('del', 'key')); + } + /** * @group disconnected */ diff --git a/tests/Predis/Connection/StreamConnectionTest.php b/tests/Predis/Connection/StreamConnectionTest.php index aa3d72c3..96b14dcf 100644 --- a/tests/Predis/Connection/StreamConnectionTest.php +++ b/tests/Predis/Connection/StreamConnectionTest.php @@ -21,6 +21,7 @@ use Predis\Connection\Resource\StreamFactoryInterface; use Predis\Consumer\Push\PushResponse; use Predis\Protocol\ProtocolException; use Predis\Response\Error as ErrorResponse; +use Predis\Response\Status; use Psr\Http\Message\StreamInterface; use RuntimeException; @@ -155,6 +156,110 @@ class StreamConnectionTest extends PredisConnectionTestCase $connection->connect(); } + /** + * @group disconnected + */ + public function testSessionCommandsAreRegisteredReplacedAndRemovable(): void + { + $connection = new StreamConnection(new Parameters(), $this->mockStreamFactory); + $prepare1 = new RawCommand('HIMPORT', ['PREPARE', 'fs1', 'a']); + $prepare1Replaced = new RawCommand('HIMPORT', ['PREPARE', 'fs1', 'a', 'b']); + $prepare2 = new RawCommand('HIMPORT', ['PREPARE', 'fs2', 'c']); + + $connection->addSessionCommand('himport:fs1', $prepare1); + $connection->addSessionCommand('himport:fs2', $prepare2); + $this->assertSame( + ['himport:fs1' => $prepare1, 'himport:fs2' => $prepare2], + $connection->getSessionCommands() + ); + + // Last registration under a key wins (mirrors server silent-replace). + $connection->addSessionCommand('himport:fs1', $prepare1Replaced); + $this->assertSame($prepare1Replaced, $connection->getSessionCommands()['himport:fs1']); + + $connection->removeSessionCommand('himport:fs1'); + $this->assertSame(['himport:fs2' => $prepare2], $connection->getSessionCommands()); + } + + /** + * @group disconnected + */ + public function testRemoveSessionCommandsByPrefix(): void + { + $connection = new StreamConnection(new Parameters(), $this->mockStreamFactory); + $connection->addSessionCommand('himport:fs1', new RawCommand('HIMPORT', ['PREPARE', 'fs1', 'a'])); + $connection->addSessionCommand('himport:fs2', new RawCommand('HIMPORT', ['PREPARE', 'fs2', 'b'])); + $connection->addSessionCommand('other:key', new RawCommand('PING')); + + $connection->removeSessionCommandsByPrefix('himport:'); + + $this->assertSame(['other:key'], array_keys($connection->getSessionCommands())); + } + + /** + * @group disconnected + */ + public function testSessionCommandsAreNotSerialized(): void + { + $connection = new StreamConnection(new Parameters(), $this->mockStreamFactory); + $connection->addSessionCommand('himport:fs1', new RawCommand('HIMPORT', ['PREPARE', 'fs1', 'a'])); + + $this->assertNotContains('sessionCommands', $connection->__sleep()); + } + + /** + * @group disconnected + */ + public function testReplaysSessionCommandsOnConnect(): void + { + $prepare = new RawCommand('HIMPORT', ['PREPARE', 'fs1', 'a', 'b']); + + $connection = $this->getMockBuilder(StreamConnection::class) + ->onlyMethods(['createResource', 'write', 'read']) + ->setConstructorArgs([new Parameters()]) + ->getMock(); + $connection->method('createResource')->willReturn($this->mockStream); + $connection + ->expects($this->once()) + ->method('write') + ->with($prepare->serializeCommand()); + $connection + ->expects($this->once()) + ->method('read') + ->willReturn(Status::get('OK')); + + $connection->addSessionCommand('himport:fs1', $prepare); + $connection->connect(); + + // Kept for future reconnects after a successful replay. + $this->assertArrayHasKey('himport:fs1', $connection->getSessionCommands()); + } + + /** + * @group disconnected + */ + public function testDropsSessionCommandOnErrorReplyWithoutTearingDownConnection(): void + { + $prepare = new RawCommand('HIMPORT', ['PREPARE', 'fs1', 'a']); + + $connection = $this->getMockBuilder(StreamConnection::class) + ->onlyMethods(['createResource', 'write', 'read']) + ->setConstructorArgs([new Parameters()]) + ->getMock(); + $connection->method('createResource')->willReturn($this->mockStream); + $connection->method('write'); + $connection + ->method('read') + ->willReturn(new ErrorResponse('ERR no such fieldset')); + + $connection->addSessionCommand('himport:fs1', $prepare); + $connection->connect(); + + // The failed session command is dropped, but the connection stays usable. + $this->assertTrue($connection->isConnected()); + $this->assertArrayNotHasKey('himport:fs1', $connection->getSessionCommands()); + } + /** * @group disconnected */ diff --git a/tests/Predis/Retry/RetryTest.php b/tests/Predis/Retry/RetryTest.php index 658edec8..03b2ea92 100644 --- a/tests/Predis/Retry/RetryTest.php +++ b/tests/Predis/Retry/RetryTest.php @@ -20,11 +20,32 @@ use Predis\Retry\Strategy\EqualBackoff; use Predis\Retry\Strategy\ExponentialBackoff; use Predis\Retry\Strategy\NoBackoff; use Predis\Retry\Strategy\RetryStrategyInterface; +use ReflectionObject; use RuntimeException; use Throwable; class RetryTest extends TestCase { + /** + * @group disconnected + */ + public function testUpdateCatchableExceptionsDoesNotAppendDuplicates(): void + { + $retry = new Retry(new NoBackoff(), 0, [ConnectionException::class]); + + // Repeated registration of the same class (as aggregate connections and + // the himport container do, once per command) must not grow the list. + $retry->updateCatchableExceptions([StreamInitException::class]); + $retry->updateCatchableExceptions([StreamInitException::class]); + $retry->updateCatchableExceptions([StreamInitException::class, ConnectionException::class]); + + $property = (new ReflectionObject($retry))->getProperty('catchableExceptions'); + $property->setAccessible(true); + $catchable = $property->getValue($retry); + + $this->assertSame([ConnectionException::class, StreamInitException::class], $catchable); + } + /** * @group disconnected * @dataProvider strategyProvider