Fix lz4 decompression for protocol v3 and v4

Author: MichaelRoosz

CHANGELOG.md

@@ -1,3 +1,8 @@
+## v1.2.0
+
+### Fixed
+* Fixed lz4 decompression for protocol v3 and v4
+
 ## v1.1.0
 
 This release introduces the ability to override the default pool of allowed protocol versions (v5, v4, v3) via the new `allowedProtocolVersions` property in `ConnectionOptions`. This is a low-level feature intended for advanced use cases and should not be used by most users. The default behavior, which attempts to negotiate the highest supported version (v5 > v4 > v3), is recommended for the majority of situations.

src/Connection.php

@@ -1106,13 +1106,18 @@ private function configureStartupOptions(Response\Supported $supportedReponse):
         $protocolVersion = ProtocolVersion::getHighestSupportedVersion($versionsSupportedByServer, $this->options->allowedProtocolVersions);
         if ($protocolVersion === null) {
 
+            $versionsSupportedByServerInOptionFormat = array_map(
+                fn (ProtocolVersion $v) => $v->inOptionFormat(),
+                $versionsSupportedByServer
+            );
+
             $allowedProtocolVersionsInOptionFormat = array_map(
                 fn (ProtocolVersion $v) => $v->inOptionFormat(),
                 $this->options->allowedProtocolVersions
             );
 
             throw new ConnectionException('Server does not support a compatible protocol version.', ExceptionCode::CONNECTION_SERVER_PROTOCOL_UNSUPPORTED->value, [
-                'proocol_versions_supported_by_server' => $serverOptions['PROTOCOL_VERSIONS'] ?? null,
+                'proocol_versions_supported_by_server' => $versionsSupportedByServerInOptionFormat,
                 'proocol_versions_supported_by_client' => ProtocolVersion::CASES_IN_OPTION_FORMAT,
                 'proocol_versions_allowed_by_connection_options' => $allowedProtocolVersionsInOptionFormat,
             ]);

src/Connection/ConnectionOptions.php

@@ -4,10 +4,15 @@
 
 namespace Cassandra\Connection;
 
+use Cassandra\Exception\ConnectionException;
+use Cassandra\Exception\ExceptionCode;
 use Cassandra\Protocol\ProtocolVersion;
 use Cassandra\ReleaseConstants;
 
 final class ConnectionOptions {
+    /**
+     * @throws \Cassandra\Exception\ConnectionException
+     */
     public function __construct(
         public readonly bool $enableCompression = false,
         public readonly bool $throwOnOverload = false,
@@ -19,6 +24,16 @@ public function __construct(
         public readonly ProtocolVersion $initialProtocolVersion = ProtocolVersion::V3,
     ) {
 
+        if (!in_array($this->initialProtocolVersion, $this->allowedProtocolVersions, true)) {
+            throw new ConnectionException(
+                'The initial protocol version must be one of the allowed protocol versions.',
+                ExceptionCode::CONNECTION_INITIAL_PROTOCOL_VERSION_NOT_IN_ALLOWED_VERSIONS->value,
+                [
+                    'initialProtocolVersion' => $this->initialProtocolVersion,
+                    'allowedProtocolVersions' => $this->allowedProtocolVersions,
+                ]
+            );
+        }
     }
 
     /**

src/Connection/ResponseReader.php

@@ -63,8 +63,28 @@ public function readResponse(Node $node, ProtocolVersion $version, bool $waitFor
             && $header->length > 0
             && $header->flags & Flag::COMPRESSION
         ) {
-            $this->lz4Decompressor->setInput($body);
+            $uncompressedLength = unpack('N', substr($body, 0, 4));
+            if ($uncompressedLength === false) {
+                throw new ConnectionException(
+                    'Cannot read uncompressed length from compressed frame',
+                    ExceptionCode::CONNECTION_CANNOT_READ_DECOMPRESSED_FRAME_LENGTH->value,
+                    []
+                );
+            }
+
+            $this->lz4Decompressor->setInput(substr($body, 4));
             $body = $this->lz4Decompressor->decompressBlock();
+
+            if ($uncompressedLength[1] !== strlen($body)) {
+                throw new ConnectionException(
+                    'Decompressed frame length does not match expected length',
+                    ExceptionCode::CONNECTION_DECOMPRESSED_FRAME_LENGTH_MISMATCH->value,
+                    [
+                        'expected_length' => $uncompressedLength,
+                        'actual_length' => strlen($body),
+                    ]
+                );
+            }
         }
 
         return $this->createResponse($header, $body);

src/Exception/ExceptionCode.php

@@ -8,7 +8,7 @@
  * Global enumeration of all exception codes used throughout the Cassandra library.
  * Each exception class has its own prefix based on its namespace or class name in uppercase.
  * 
- * Next free code: 1323
+ * Next free code: 1326
  */
 enum ExceptionCode: int {
     // CompressionException (COMPRESSION_)
@@ -30,9 +30,12 @@ enum ExceptionCode: int {
     case CONNECTION_AUTO_PREPARE_UNEXPECTED_RESPONSE = 1306;
     case CONNECTION_AUTO_PREPARE_UNEXPECTED_RESPONSE_REEXECUTE = 1304;
     case CONNECTION_AUTO_PREPARE_UNEXPECTED_RESULT_TYPE = 1302;
+    case CONNECTION_CANNOT_READ_DECOMPRESSED_FRAME_LENGTH = 1324;
     case CONNECTION_CANNOT_READ_RESPONSE_HEADER = 1008;
     case CONNECTION_COMPRESSION_NOT_SUPPORTED = 1009;
+    case CONNECTION_DECOMPRESSED_FRAME_LENGTH_MISMATCH = 1323;
     case CONNECTION_EXECUTE_UNEXPECTED_RESPONSE = 1010;
+    case CONNECTION_INITIAL_PROTOCOL_VERSION_NOT_IN_ALLOWED_VERSIONS = 1325;
     case CONNECTION_INVALID_OPCODE_TYPE = 1012;
     case CONNECTION_NOT_CONNECTED = 1013;
     case CONNECTION_OPTIONS_UNEXPECTED_RESPONSE = 1014;

test/Integration/CompressionTest.php

@@ -8,11 +8,46 @@
 use Cassandra\Connection\ConnectionOptions;
 use Cassandra\Connection\SocketNodeConfig;
 use Cassandra\Consistency;
+use Cassandra\Protocol\ProtocolVersion;
 use Cassandra\Request\Options\QueryOptions;
 
 final class CompressionTest extends AbstractIntegrationTestCase {
-    // todo: test with v3,v4,v5 protocol versions
-    public function testNegotiatesCompressionAndDecodesCompressedFrames(): void {
+    public function testNegotiatesCompressionAndDecodesCompressedFramesV3(): void {
+
+        $options = new ConnectionOptions(
+            enableCompression: true,
+            allowedProtocolVersions: [ProtocolVersion::V3],
+        );
+
+        $this->testCompression($options);
+    }
+
+    public function testNegotiatesCompressionAndDecodesCompressedFramesV4(): void {
+
+        $options = new ConnectionOptions(
+            enableCompression: true,
+            allowedProtocolVersions: [ProtocolVersion::V4],
+            initialProtocolVersion: self::isScyllaDb() ? ProtocolVersion::V4 : ProtocolVersion::V3,
+        );
+
+        $this->testCompression($options);
+    }
+
+    public function testNegotiatesCompressionAndDecodesCompressedFramesV5(): void {
+
+        if (self::isScyllaDb()) {
+            $this->markTestSkipped('Skipping V5 compression test on ScyllaDB');
+        }
+
+        $options = new ConnectionOptions(
+            enableCompression: true,
+            allowedProtocolVersions: [ProtocolVersion::V5],
+        );
+
+        $this->testCompression($options);
+    }
+
+    private function testCompression(ConnectionOptions $options): void {
 
         $nodes = [
             new SocketNodeConfig(
@@ -23,7 +58,6 @@ public function testNegotiatesCompressionAndDecodesCompressedFrames(): void {
             ),
         ];
 
-        $options = new ConnectionOptions(enableCompression: true);
         $conn = new Connection($nodes, self::$defaultKeyspace, $options);
         $conn->setConsistency(Consistency::ONE);
         $conn->connect();