prepared-metadata: harden cached decoder state

Author: dkropachev

CHANGELOG.rst

@@ -7,13 +7,26 @@ Features
   statements skip re-sending result metadata on EXECUTE, and the driver automatically
   refreshes cached metadata when the server detects a schema change (DRIVER-153)
 
+Bug Fixes
+---------
+* Harden prepared-result metadata caching against concurrent schema/client decoder
+  changes. Cached metadata, ids, and decoder provenance are published atomically;
+  UDT descriptors are isolated by registered Python class; UDT registration
+  invalidates cached ids before publishing mapping changes; and futures refresh
+  their request snapshot before fetching another page or after re-prepare.
+* Restrict ``skip_meta`` to immutable snapshots of built-in protocol handlers.
+  Modifying their public decoder configuration disables the optimization until
+  it is restored, while custom handlers continue to receive full result metadata.
+
 Others
 ------
 * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are
   now read-only. They are replaced together by
   ``PreparedStatement.update_result_metadata()``, so a request can never observe a metadata
   id paired with result metadata from a different schema version. Code that assigned either
-  attribute directly must call ``update_result_metadata()`` instead.
+  attribute directly must call ``update_result_metadata()`` instead. The metadata getter
+  keeps its historical list shape but returns a defensive copy, so mutating it cannot
+  change the internal metadata/id pair.
 * Message serialization now receives the connection's negotiated ``ProtocolFeatures``:
   ``Connection.send_msg`` passes ``protocol_features`` to the encoder, and
   ``_ProtocolHandler.encode_message`` forwards it to each message's ``send_body``.
@@ -24,8 +37,10 @@ Others
   forward it. There is deliberately no compatibility fallback: protocol extensions
   are negotiated per connection at STARTUP, so an encoder unaware of
   ``protocol_features`` could silently omit fields a negotiated extension requires.
-  This release emits no new bytes on the wire; the parameter is groundwork for
-  upcoming protocol extensions (``SCYLLA_USE_METADATA_ID``, ``TABLETS_ROUTING_V2``).
+  When ``SCYLLA_USE_METADATA_ID`` is negotiated, EXECUTE messages use this plumbing
+  to include the extension's metadata id and may request that result metadata be
+  omitted. Connections that do not negotiate the extension retain their existing
+  frame layout.
 
 3.29.11
 =======

cassandra/cluster.py

@@ -41,6 +41,7 @@
 import time
 import struct
 import sys
+from threading import Lock
 from uuid import UUID
 
 from cassandra.marshal import (int8_pack, int8_unpack, int16_pack, int16_unpack,
@@ -977,30 +978,49 @@ class UserType(TupleType):
     typename = "org.apache.cassandra.db.marshal.UserType"
 
     _cache = {}
+    _cache_lock = Lock()
     _module = sys.modules[__name__]
 
     @classmethod
-    def make_udt_class(cls, keyspace, udt_name, field_names, field_types):
+    def make_udt_class(cls, keyspace, udt_name, field_names, field_types,
+                       mapped_class=None):
         assert len(field_names) == len(field_types)
 
-        instance = cls._cache.get((keyspace, udt_name))
-        if not instance or instance.fieldnames != field_names or instance.subtypes != field_types:
-            instance = type(udt_name, (cls,), {'subtypes': field_types,
-                                               'cassname': cls.cassname,
-                                               'typename': udt_name,
-                                               'fieldnames': field_names,
-                                               'keyspace': keyspace,
-                                               'mapped_class': None,
-                                               'tuple_type': cls._make_registered_udt_namedtuple(keyspace, udt_name, field_names)})
-            cls._cache[(keyspace, udt_name)] = instance
+        # A UDT type descriptor can outlive the response that created it (for
+        # example, in a PreparedStatement's cached result metadata).  Keep a
+        # distinct descriptor for each registered Python mapping so parsing
+        # metadata for another Cluster cannot mutate an existing descriptor.
+        #
+        # Use identity rather than the class itself in the key. Registered
+        # classes normally are hashable, but a custom metaclass may opt out of
+        # hashing. The generated descriptor holds a strong reference to
+        # mapped_class, so its id cannot be reused while this cache entry exists.
+        cache_key = (keyspace, udt_name, id(mapped_class))
+        with cls._cache_lock:
+            instance = cls._cache.get(cache_key)
+            if not instance or instance.fieldnames != field_names or instance.subtypes != field_types:
+                instance = type(udt_name, (cls,), {'subtypes': field_types,
+                                                   'cassname': cls.cassname,
+                                                   'typename': udt_name,
+                                                   'fieldnames': field_names,
+                                                   'keyspace': keyspace,
+                                                   'mapped_class': mapped_class,
+                                                   'tuple_type': cls._make_registered_udt_namedtuple(keyspace, udt_name, field_names)})
+                cls._cache[cache_key] = instance
         return instance
 
     @classmethod
     def evict_udt_class(cls, keyspace, udt_name):
-        try:
-            del cls._cache[(keyspace, udt_name)]
-        except KeyError:
-            pass
+        # Registration changes must evict mapped and unmapped variants alike.
+        # Slicing also recognizes two-element keys left by older driver code in
+        # a long-running process that reloads this module.
+        with cls._cache_lock:
+            cache_keys = [
+                cache_key for cache_key in cls._cache
+                if cache_key[:2] == (keyspace, udt_name)
+            ]
+            for cache_key in cache_keys:
+                del cls._cache[cache_key]
 
     @classmethod
     def apply_parameters(cls, subtypes, names):

cassandra/cqltypes.py

@@ -891,9 +891,9 @@ def read_type(cls, f, user_type_map):
             num_fields = read_short(f)
             names, types = zip(*((read_string(f), cls.read_type(f, user_type_map))
                                  for _ in range(num_fields)))
-            specialized_type = typeclass.make_udt_class(ks, udt_name, names, types)
-            specialized_type.mapped_class = user_type_map.get(ks, {}).get(udt_name)
-            typeclass = specialized_type
+            mapped_class = user_type_map.get(ks, {}).get(udt_name)
+            typeclass = typeclass.make_udt_class(
+                ks, udt_name, names, types, mapped_class=mapped_class)
         elif typeclass == CUSTOM_TYPE:
             classname = read_string(f)
             typeclass = lookup_casstype(classname)
@@ -1100,6 +1100,110 @@ def send_body(self, f, protocol_version, protocol_features=None):
                     "Continuous paging backpressure is not supported.")
 
 
+_PREPARED_METADATA_RESULT_ATTRIBUTES = (
+    'opcode',
+    'name',
+    'kind',
+    'results',
+    'paging_state',
+    '_FLAGS_GLOBAL_TABLES_SPEC',
+    '_HAS_MORE_PAGES_FLAG',
+    '_NO_METADATA_FLAG',
+    '_CONTINUOUS_PAGING_FLAG',
+    '_CONTINUOUS_PAGING_LAST_FLAG',
+    '_METADATA_ID_FLAG',
+    'column_names',
+    'column_types',
+    'parsed_rows',
+    'continuous_paging_seq',
+    'continuous_paging_last',
+    'new_keyspace',
+    'column_metadata',
+    'query_id',
+    'bind_metadata',
+    'pk_indexes',
+    'schema_change_event',
+    'is_lwt',
+    'tracing',
+    'custom_payload',
+    'warnings',
+    '__init__',
+    'recv',
+    'recv_body',
+    'recv_results_rows',
+    'recv_results_prepared',
+    'recv_results_metadata',
+    'recv_prepared_metadata',
+    'recv_results_schema_change',
+    'read_type',
+    'recv_row',
+)
+
+
+def _class_attribute(cls, name):
+    for base in cls.__mro__:
+        if name in vars(base):
+            return vars(base)[name]
+    return None
+
+
+_PREPARED_METADATA_CONFIG_UNSET = object()
+
+
+def _prepared_metadata_cache_config(
+        handler, result_message_type,
+        message_types=_PREPARED_METADATA_CONFIG_UNSET,
+        type_codes=_PREPARED_METADATA_CONFIG_UNSET,
+        code_to_type=_PREPARED_METADATA_CONFIG_UNSET,
+        decode_message=_PREPARED_METADATA_CONFIG_UNSET,
+        result_attributes=_PREPARED_METADATA_CONFIG_UNSET,
+        column_encryption_policy=_PREPARED_METADATA_CONFIG_UNSET):
+    """
+    Capture the canonical driver-owned result-decoder configuration.
+
+    Cluster-side snapshots compare against this exact configuration. Installing
+    an application ResultMessage or mutating a decoding attribute therefore
+    makes the handler ineligible for skip-metadata instead of attempting to
+    freeze arbitrary application class state.
+    """
+    message_types = (
+        handler.message_types_by_opcode
+        if message_types is _PREPARED_METADATA_CONFIG_UNSET
+        else message_types)
+    type_codes = (
+        result_message_type.type_codes
+        if type_codes is _PREPARED_METADATA_CONFIG_UNSET
+        else type_codes)
+    code_to_type = (
+        getattr(result_message_type, 'code_to_type', {})
+        if code_to_type is _PREPARED_METADATA_CONFIG_UNSET
+        else code_to_type)
+    decode_message = (
+        _class_attribute(handler, 'decode_message')
+        if decode_message is _PREPARED_METADATA_CONFIG_UNSET
+        else decode_message)
+    result_attributes = (
+        tuple(
+            (name, _class_attribute(result_message_type, name))
+            for name in _PREPARED_METADATA_RESULT_ATTRIBUTES
+        )
+        if result_attributes is _PREPARED_METADATA_CONFIG_UNSET
+        else result_attributes)
+    column_encryption_policy = (
+        handler.column_encryption_policy
+        if column_encryption_policy is _PREPARED_METADATA_CONFIG_UNSET
+        else column_encryption_policy)
+    return (
+        tuple(message_types.items()),
+        result_message_type,
+        tuple(type_codes.items()),
+        tuple(code_to_type.items()),
+        decode_message,
+        result_attributes,
+        column_encryption_policy,
+    )
+
+
 class _ProtocolHandler(object):
     """
     _ProtocolHander handles encoding and decoding messages.
@@ -1121,6 +1225,11 @@ class _ProtocolHandler(object):
     column_encryption_policy = None
     """Instance of :class:`cassandra.policies.ColumnEncryptionPolicy` in use by this handler"""
 
+    # Marks handlers whose result-decoder configuration the driver may copy
+    # into an immutable per-request snapshot. Subclasses do not inherit
+    # eligibility because their opcode and type maps are application-owned.
+    _prepared_metadata_cache_token = object()
+
     @classmethod
     def encode_message(cls, msg, stream_id, protocol_version, compressor, allow_beta_protocol_version,
                        protocol_features):
@@ -1245,6 +1354,10 @@ def decode_message(cls, protocol_version, protocol_features, user_type_map, stre
         return msg
 
 
+_ProtocolHandler._prepared_metadata_cache_config = \
+    _prepared_metadata_cache_config(_ProtocolHandler, ResultMessage)
+
+
 def cython_protocol_handler(colparser):
     """
     Given a column parser to deserialize ResultMessages, return a suitable
@@ -1284,7 +1397,11 @@ class CythonProtocolHandler(_ProtocolHandler):
         message_types_by_opcode = my_opcodes
 
         col_parser = colparser
+        _prepared_metadata_cache_token = object()
 
+    CythonProtocolHandler._prepared_metadata_cache_config = \
+        _prepared_metadata_cache_config(
+            CythonProtocolHandler, FastResultMessage)
     return CythonProtocolHandler