protocols: stop option, chunk and block dispatch writing to shared registries

docs/source/pcapkit/protocols/protocol.rst

@@ -61,6 +61,7 @@ utility arguments and methods of specified protocols.
    .. automethod:: _make_index
    .. automethod:: _make_payload
 
+   .. automethod:: _lookup_registry
    .. automethod:: _lookup_next_layer
    .. automethod:: _decode_next_layer
    .. automethod:: _import_next_layer

pcapkit/protocols/application/httpv2.py

@@ -225,7 +225,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_HTTP':
         if schema.type in (Enum_Frame.SETTINGS, Enum_Frame.PING) and schema.stream['sid'] != 0:
             raise ProtocolError(f'HTTP/2: [Type {schema.type}] invalid format')
 
-        name = self.__frame__[schema.type]
+        name = self._lookup_registry(self.__frame__, schema.type)
         if isinstance(name, str):
             meth_name = f'_read_http_{name}'
             meth = cast('FrameParser',
@@ -268,7 +268,7 @@ def make(self,  # type: ignore[override]
         if isinstance(frame, bytes):
             frame_val = frame  # type: bytes | Schema_FrameType
         elif isinstance(frame, (dict, Data_HTTP)):
-            name = self.__frame__[type_val]
+            name = self._lookup_registry(self.__frame__, type_val)
             if isinstance(name, str):
                 meth_name = f'_make_http_{name}'
                 meth = cast('FrameConstructor',

pcapkit/protocols/internet/hopopt.py

@@ -467,7 +467,7 @@ def _read_hopopt_options(self, length: 'int') -> 'Option':
 
         for schema in self.__header__.options:
             dscp = schema.type
-            name = self.__option__[dscp]
+            name = self._lookup_registry(self.__option__, dscp)
 
             if isinstance(name, str):
                 meth_name = f'_read_opt_{name}'
@@ -1275,7 +1275,7 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option,
                     if code in (Enum_Option.Pad1, Enum_Option.PadN):  # ignore padding options by default
                         continue
 
-                    name = self.__option__[code]  # type: str | tuple[OptionParser, OptionConstructor]
+                    name = self._lookup_registry(self.__option__, code)
                     if isinstance(name, str):
                         meth_name = f'_make_opt_{name}'
                         meth = cast('OptionConstructor',
@@ -1306,7 +1306,7 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option,
             if code in (Enum_Option.Pad1, Enum_Option.PadN):
                 continue
 
-            name = self.__option__[code]
+            name = self._lookup_registry(self.__option__, code)
             if isinstance(name, str):
                 meth_name = f'_make_opt_{name}'
                 meth = cast('OptionConstructor',

pcapkit/protocols/internet/ipv6_opts.py

@@ -478,7 +478,7 @@ def _read_ipv6_opts(self, length: 'int') -> 'Option':
 
         for schema in self.__header__.options:
             dscp = schema.type
-            name = self.__option__[dscp]
+            name = self._lookup_registry(self.__option__, dscp)
 
             if isinstance(name, str):
                 meth_name = f'_read_opt_{name}'
@@ -1287,7 +1287,7 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict
                     if code in (Enum_Option.Pad1, Enum_Option.PadN):  # ignore padding options by default
                         continue
 
-                    name = self.__option__[code]  # type: str | tuple[OptionParser, OptionConstructor]
+                    name = self._lookup_registry(self.__option__, code)
                     if isinstance(name, str):
                         meth_name = f'_make_opt_{name}'
                         meth = cast('OptionConstructor',
@@ -1318,7 +1318,7 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict
             if code in (Enum_Option.Pad1, Enum_Option.PadN):
                 continue
 
-            name = self.__option__[code]
+            name = self._lookup_registry(self.__option__, code)
             if isinstance(name, str):
                 meth_name = f'_make_opt_{name}'
                 meth = cast('OptionConstructor',

pcapkit/protocols/internet/ipv6_route.py

@@ -201,7 +201,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False,  #
             length = len(self)
         schema = self.__header__
 
-        name = self.__routing__[schema.type]
+        name = self._lookup_registry(self.__routing__, schema.type)
         if isinstance(name, str):
             name = f'_read_data_type_{name.lower()}'
             meth = cast('TypeParser',
@@ -260,7 +260,7 @@ def make(self,
             length = math.ceil((len(data) + 4) / 8)
             data_val = data.ljust(length * 8 - 4, b'\x00')  # type: bytes | Schema_RoutingType
         elif isinstance(data, (dict, Data_IPv6_Route)):
-            name = self.__routing__[type_val]
+            name = self._lookup_registry(self.__routing__, type_val)
             if isinstance(name, str):
                 name = f'_make_data_type_{name.lower()}'
                 meth = cast('TypeConstructor',

pcapkit/protocols/internet/mh.py

@@ -675,7 +675,7 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4,
             length = len(self)
         schema = self.__header__
 
-        name = self.__message__[schema.type]
+        name = self._lookup_registry(self.__message__, schema.type)
         if isinstance(name, str):
             meth_name = f'_read_msg_{name}'
             meth = cast('PacketParser',
@@ -729,7 +729,7 @@ def make(self,
         if isinstance(data, bytes):
             data_val = data  # type: bytes | Schema_Packet
         elif isinstance(data, (dict, Data_MH)):
-            name = self.__message__[type_val]
+            name = self._lookup_registry(self.__message__, type_val)
             if isinstance(name, str):
                 meth_name = f'_make_msg_{name}'
                 meth = cast('PacketConstructor',
@@ -1533,7 +1533,7 @@ def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option':
 
         for schema in options_schema:
             type = schema.type
-            name = self.__option__[type]
+            name = self._lookup_registry(self.__option__, type)
 
             if isinstance(name, str):
                 meth_name = f'_read_opt_{name}'
@@ -2337,7 +2337,7 @@ def _read_cga_extensions(self, extensions_schema: 'list[Schema_CGAExtension]') -
 
         for schema in extensions_schema:
             type = schema.type
-            name = self.__extension__[type]
+            name = self._lookup_registry(self.__extension__, type)
 
             if isinstance(name, str):
                 meth_name = f'_read_ext_{name}'
@@ -3111,7 +3111,7 @@ def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Op
                     data = schema
                 else:
                     code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema)
-                    name = self.__option__[code]
+                    name = self._lookup_registry(self.__option__, code)
                     if isinstance(name, str):
                         meth_name = f'_make_opt_{name}'
                         meth = cast('OptionConstructor',
@@ -3125,7 +3125,7 @@ def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Op
 
         options_list = []
         for code, option in options.items(multi=True):
-            name = self.__option__[code]
+            name = self._lookup_registry(self.__option__, code)
             if isinstance(name, str):
                 meth_name = f'_make_opt_{name}'
                 meth = cast('OptionConstructor',
@@ -3780,7 +3780,7 @@ def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension
                     data_len = len(schema.pack())
                 else:
                     code, args = cast('tuple[Enum_CGAExtension, dict[str, Any]]', schema)
-                    name = self.__extension__[code]
+                    name = self._lookup_registry(self.__extension__, code)
                     if isinstance(name, str):
                         meth_name = f'_make_ext_{name}'
                         meth = cast('ExtensionConstructor',
@@ -3797,7 +3797,7 @@ def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension
 
         extensions_list = []
         for code, extension in extensions.items(multi=True):
-            name = self.__extension__[code]
+            name = self._lookup_registry(self.__extension__, code)
             if isinstance(name, str):
                 meth_name = f'_make_ext_{name}'
                 meth = cast('ExtensionConstructor',

pcapkit/protocols/misc/pcapng.py

@@ -950,7 +950,7 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True,
             self._sect += 1
             self._ctx = None
 
-        name = self.__block__[schema.type]
+        name = self._lookup_registry(self.__block__, schema.type)
         if isinstance(name, str):
             meth_name = f'_read_block_{name}'
             meth = cast('BlockParser',
@@ -1010,7 +1010,7 @@ def make(self,
         if isinstance(block, bytes):
             block_val = block  # type: bytes | Schema_BlockType
         elif isinstance(block, (dict, Data_PCAPNG)):
-            name = self.__block__[type_val]
+            name = self._lookup_registry(self.__block__, type_val)
             if isinstance(name, str):
                 meth_name = f'_make_block_{name}'
                 meth = cast('BlockConstructor',
@@ -1876,7 +1876,7 @@ def _read_block_dsb(self, schema: 'Schema_DecryptionSecretsBlock', *,
             Parsed packet data.
 
         """
-        name = self.__secrets__[schema.secrets_type]
+        name = self._lookup_registry(self.__secrets__, schema.secrets_type)
         if isinstance(name, str):
             meth_name = f'_read_secrets_{name}'
             meth = cast('SecretsParser',
@@ -2040,7 +2040,7 @@ def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option
 
         for schema in options_schema:
             type = schema.type
-            name = self.__option__[_option_key(type)]
+            name = self._lookup_registry(self.__option__, _option_key(type))
 
             if isinstance(name, str):
                 meth_name = f'_read_option_{name}'
@@ -3126,7 +3126,7 @@ def _read_nrb_records(self, records_schema: 'list[Schema_NameResolutionRecord]')
 
         for schema in records_schema:
             type = schema.type
-            name = self.__record__[type]
+            name = self._lookup_registry(self.__record__, type)
 
             if isinstance(name, str):
                 meth_name = f'_read_record_{name}'
@@ -3736,7 +3736,7 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None,
         if isinstance(secrets_data, bytes):
             secrets_data_val = secrets_data  # type: bytes | Schema_DSBSecrets
         elif isinstance(secrets_data, (dict, Data_DSBSecrets)):
-            name = self.__secrets__[secrets_type_val]
+            name = self._lookup_registry(self.__secrets__, secrets_type_val)
             if isinstance(name, str):
                 meth_name = f'_make_secrets_{name}'
                 meth = cast('SecretsConstructor',
@@ -3925,7 +3925,7 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu
                         has_endofopt = True
                         continue
 
-                    name = self.__option__[_option_key(code)]
+                    name = self._lookup_registry(self.__option__, _option_key(code))
                     if isinstance(name, str):
                         meth_name = f'_make_option_{name}'
                         meth = cast('OptionConstructor',
@@ -3952,7 +3952,7 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu
                 has_endofopt = True
                 continue
 
-            name = self.__option__[_option_key(code)]
+            name = self._lookup_registry(self.__option__, _option_key(code))
             if isinstance(name, str):
                 meth_name = f'_make_option_{name}'
                 meth = cast('OptionConstructor',
@@ -5353,7 +5353,7 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord
                         has_record_end = True
                         continue
 
-                    name = self.__record__[code]
+                    name = self._lookup_registry(self.__record__, code)
                     if isinstance(name, str):
                         meth_name = f'_make_record_{name}'
                         meth = cast('RecordConstructor',
@@ -5379,7 +5379,7 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord
                 has_record_end = True
                 continue
 
-            name = self.__record__[code]
+            name = self._lookup_registry(self.__record__, code)
             if isinstance(name, str):
                 meth_name = f'_make_record_{name}'
                 meth = cast('RecordConstructor',

pcapkit/protocols/protocol.py

@@ -58,6 +58,7 @@
 _PT = TypeVar('_PT', bound='Data')
 _ST = TypeVar('_ST', bound='Schema')
 _CTX = TypeVar('_CTX', bound='ProtocolContext')
+_VT = TypeVar('_VT')
 
 # readable characters' order list
 readable = [ord(char) for char in filter(lambda char: not char.isspace(), string.printable)]
@@ -1237,6 +1238,42 @@ def _make_payload(cls, data: 'Data') -> 'ProtocolBase':
 
         return proto.from_data(data[name])
 
+    @staticmethod
+    def _lookup_registry(registry: 'DefaultDict[Any, _VT]', code: 'Any') -> '_VT':
+        """Look up a dispatch registry entry without recording a miss.
+
+        Arguments:
+            registry: dispatch registry to read, i.e. :attr:`self.__proto__
+                <ProtocolBase.__proto__>` or one of the per-protocol
+                ``__option__`` / ``__chunk__`` / ``__block__`` family. Passed in
+                rather than read from the class, so that a caller reaching the
+                registry through an instance keeps doing so.
+            code: registry key to look up, i.e. the wire code being dispatched on
+
+        Returns:
+            The entry registered for ``code``, or the fallback ``registry``
+            declares when ``code`` is not registered.
+
+        Important:
+            Every one of these registries is a :class:`collections.defaultdict`
+            held on a *class* attribute, shared by every instance of the class in
+            the process. So ``registry[code]`` inserts each code it misses, and
+            parsing one packet carrying an unrecognised code is enough to grow
+            the registry permanently.
+
+            The inserted value is whatever the default factory would have
+            produced anyway, so the entry buys nothing. It costs a spurious
+            "already registered" warning from the next genuine ``register`` call
+            for that code, and it makes "is this code registered?"
+            unanswerable by inspection, since the answer depends on what has
+            been parsed. The fallback is therefore read from the default factory
+            directly rather than through a lookup that records it.
+
+        """
+        if code in registry:
+            return registry[code]
+        return cast('Callable[[], _VT]', registry.default_factory)()
+
     @staticmethod
     def _lookup_next_layer(registry: 'DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]',
                            proto: 'int') -> 'Type[ProtocolBase]':
@@ -1255,31 +1292,29 @@ def _lookup_next_layer(registry: 'DefaultDict[int, ModuleDescriptor[ProtocolBase
             ``proto`` is not registered.
 
         Important:
-            ``registry`` is a :class:`collections.defaultdict`, so indexing it
-            with an unregistered code would *insert* that code. It is
-            class-level -- shared by every instance in the process -- so parsing
-            a single packet with an unregistered code would grow it, and make
-            :meth:`self.register <ProtocolBase.register>` afterwards report that
-            code as already registered. The fallback is therefore read from the
-            default factory rather than through a lookup that records it.
-
-            Resolving a :class:`~pcapkit.corekit.module.ModuleDescriptor` is
-            still written back, since that is memoisation of an import for a
-            code that *is* registered rather than a new entry.
+            The lookup itself is :meth:`self._lookup_registry
+            <ProtocolBase._lookup_registry>`, so a miss does not grow the shared
+            registry. What this adds is the next-layer-specific resolution step:
+            a registered code may hold a
+            :class:`~pcapkit.corekit.module.ModuleDescriptor` rather than a
+            class, and importing it is written back so the import happens once.
+
+            That write-back is deliberately confined to a *hit*. Memoising the
+            fallback's resolution under ``proto`` would be exactly the insertion
+            :meth:`self._lookup_registry <ProtocolBase._lookup_registry>` exists
+            to avoid.
 
         """
-        if proto in registry:
-            protocol = registry[proto]
-            if isinstance(protocol, ModuleDescriptor):
-                protocol = protocol.klass
-                registry[proto] = protocol  # update mapping upon import
-            return protocol
-
-        fallback = cast('Callable[[], ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]',
-                        registry.default_factory)()
-        if isinstance(fallback, ModuleDescriptor):
-            return fallback.klass
-        return fallback
+        protocol = ProtocolBase._lookup_registry(registry, proto)
+        if isinstance(protocol, ModuleDescriptor):
+            klass = protocol.klass
+            # a descriptor can also come back from the default factory, and that
+            # one has no key to memoise under -- writing it back would recreate
+            # the insertion-on-miss this exists to avoid
+            if proto in registry:
+                registry[proto] = klass  # update mapping upon import
+            return klass
+        return protocol
 
     def _decode_next_layer(self, dict_: '_PT', proto: 'int', length: 'Optional[int]' = None, *,
                            packet: 'Optional[dict[str, Any]]' = None) -> '_PT':