[packet_interface] Check at validation that fragments fit the reliability queue

- Work out each layer's packet size during validation, as the C++ does,
  and reject a fragmentation max_packet_size whose fragments do not all
  fit in the reliability queue at once, or need more than 128 fragments
- Fit the test peer's largest packet in its queue
- Reuse interface_layers() in final_validate_interface()
- Keep the earlier tests in place in test_packet_interface.py

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

esphome/components/packet_interface/__init__.py

@@ -22,6 +22,7 @@ from .layers import (
     STACK_ORDER,
     PacketInterface,
     build_stack,
+    check_packet_sizes,
     check_session,
     check_single_peer,
     packet_interface_ns,
@@ -74,8 +75,11 @@ def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]:
 def _validate_layers(config: ConfigType) -> ConfigType:
     """With peers, the layers go in each peer; otherwise check them against the one destination."""
     check_session(config)
+    size = config[CONF_MAX_PACKET_SIZE]
     if (addressing := config.get(CONF_ADDRESSING)) is None:
+        check_packet_sizes(config, size)
         return config
+    size = ADDRESSING.get_type(addressing).packet_size(addressing, size)
     if CONF_PEERS in addressing:
         for kind in STACK_ORDER:
             if kind.key in config:
@@ -83,9 +87,16 @@ def _validate_layers(config: ConfigType) -> ConfigType:
                     f"With '{CONF_PEERS}', '{kind.key}' goes in each peer",
                     path=[kind.key],
                 )
+        for index, peer in enumerate(addressing[CONF_PEERS]):
+            try:
+                check_packet_sizes(peer, size)
+            except cv.Invalid as err:
+                err.prepend([CONF_ADDRESSING, CONF_PEERS, index])
+                raise
         return config
     addressing.setdefault(CONF_DESTINATION, BROADCAST_ADDRESS)
     check_single_peer(config, addressing[CONF_DESTINATION])
+    check_packet_sizes(config, size)
     return config
 
 
@@ -108,14 +119,12 @@ def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]:
 
     def validator(config: ConfigType) -> ConfigType:
         interface_id: ID = config[key]
-        for interface in fv.full_config.get().get(DOMAIN, []):
-            if interface[CONF_ID].id == interface_id.id and CONF_PEERS in interface.get(
-                CONF_ADDRESSING, {}
-            ):
-                raise cv.Invalid(
-                    f"'{interface_id.id}' has several peers, so use the id of one of them",
-                    path=[key],
-                )
+        interface = interface_layers(interface_id)
+        if interface is not None and CONF_PEERS in interface.get(CONF_ADDRESSING, {}):
+            raise cv.Invalid(
+                f"'{interface_id.id}' has several peers, so use the id of one of them",
+                path=[key],
+            )
         return config
 
     return validator

esphome/components/packet_interface/layers/__init__.py

@@ -44,6 +44,18 @@ def _nothing(config: ConfigType) -> list[str]:
     return []
 
 
+def _same_size(config: ConfigType, size: int) -> int:
+    return size
+
+
+def _no_limit(config: ConfigType) -> int | None:
+    return None
+
+
+def _fits(config: ConfigType, size: int, capacity: int | None) -> None:
+    """Accept any layer below."""
+
+
 @dataclass(frozen=True)
 class LayerType:
     """One implementation of a kind of layer."""
@@ -57,6 +69,12 @@ class LayerType:
     needs_session: Callable[[ConfigType], bool]
     # The components this configuration needs loaded
     auto_load: Callable[[ConfigType], list[str]]
+    # The largest packet this configuration takes, from the largest that the layer below takes, as the C++ works it out
+    packet_size: Callable[[ConfigType, int], int]
+    # How many packets this configuration takes at once, or None for no limit
+    send_capacity: Callable[[ConfigType], int | None]
+    # Raises cv.Invalid if this configuration cannot work above a layer with this packet size and send capacity
+    check_lower: Callable[[ConfigType, int, int | None], None]
 
 
 class LayerKind:
@@ -82,6 +100,9 @@ class LayerKind:
         single_peer: Callable[[ConfigType], bool] = _never,
         needs_session: Callable[[ConfigType], bool] = _never,
         auto_load: Callable[[ConfigType], list[str]] = _nothing,
+        packet_size: Callable[[ConfigType, int], int] = _same_size,
+        send_capacity: Callable[[ConfigType], int | None] = _no_limit,
+        check_lower: Callable[[ConfigType, int, int | None], None] = _fits,
     ) -> Callable[[LayerBuilder], LayerBuilder]:
         """Register a type with its schema, or with a function returning a schema that contains other layers."""
 
@@ -104,6 +125,9 @@ class LayerKind:
                 single_peer,
                 needs_session,
                 auto_load,
+                packet_size,
+                send_capacity,
+                check_lower,
             )
             return build
 
@@ -178,6 +202,26 @@ def check_single_peer(config: ConfigType, destination: int) -> None:
             )
 
 
+def remaining_packet_size(size: int, used: int) -> int:
+    return max(size - used, 0)
+
+
+def check_packet_sizes(config: ConfigType, size: int) -> None:
+    """Check each layer in config against the layer below it; size is the largest packet that the stack sits on takes."""
+    capacity = None
+    for kind in STACK_ORDER:
+        if (layer := config.get(kind.key)) is None:
+            continue
+        layer_type = kind.get_type(layer)
+        try:
+            layer_type.check_lower(layer, size, capacity)
+        except cv.Invalid as err:
+            err.prepend([kind.key])
+            raise
+        size = layer_type.packet_size(layer, size)
+        capacity = layer_type.send_capacity(layer)
+
+
 def stack_auto_load(config: ConfigType) -> set[str]:
     """The components that the layers configured in config need loaded."""
     loads: set[str] = set()

esphome/components/packet_interface/layers/basic_addressing.py

@@ -18,11 +18,15 @@ from . import (
     check_session,
     check_single_peer,
     packet_interface_ns,
+    remaining_packet_size,
     stack_schema,
 )
 
 CONF_PORT_ID = "port_id"
 
+# The destination and source addresses
+HEADER_SIZE = 4
+
 BasicAddressing = packet_interface_ns.class_("BasicAddressing", cg.Component)
 AddressingPort = packet_interface_ns.class_("AddressingPort", PacketInterface)
 
@@ -103,7 +107,11 @@ async def _new_port(
     return port
 
 
-@ADDRESSING.register("basic", schema_factory=_basic_addressing_schema)
+@ADDRESSING.register(
+    "basic",
+    schema_factory=_basic_addressing_schema,
+    packet_size=lambda config, size: remaining_packet_size(size, HEADER_SIZE),
+)
 async def build_basic_addressing(
     config: ConfigType, link: MockObj
 ) -> list[tuple[MockObj, ConfigType | None]]:

esphome/components/packet_interface/layers/basic_fragmentation.py

@@ -1,5 +1,7 @@
 """Fragmentation layer that splits packets larger than the layers below allow, and joins them again."""
 
+import math
+
 import esphome.codegen as cg
 from esphome.components.const import CONF_MAX_PACKET_SIZE
 import esphome.config_validation as cv
@@ -9,6 +11,10 @@ from esphome.types import ConfigType
 
 from . import FRAGMENTATION, PacketInterface, packet_interface_ns
 
+# The packet number, and the fragment's index with the flag on the last fragment
+HEADER_SIZE = 2
+MAX_FRAGMENTS = 128
+
 BasicFragmentation = packet_interface_ns.class_("BasicFragmentation", PacketInterface)
 
 BASIC_FRAGMENTATION_SCHEMA = cv.Schema(
@@ -20,9 +26,29 @@ BASIC_FRAGMENTATION_SCHEMA = cv.Schema(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
+def _check_lower(config: ConfigType, size: int, capacity: int | None) -> None:
+    """The largest packet must fit in the fragments that the layer below can take at once."""
+    fragment_size = size - HEADER_SIZE
+    if fragment_size <= 0:
+        raise cv.Invalid("The layers below leave no room for fragments")
+    fragments = max(1, math.ceil(config[CONF_MAX_PACKET_SIZE] / fragment_size))
+    limit = MAX_FRAGMENTS if capacity is None else min(MAX_FRAGMENTS, capacity)
+    if fragments > limit:
+        raise cv.Invalid(
+            f"A packet of {config[CONF_MAX_PACKET_SIZE]} bytes needs {fragments} fragments of {fragment_size} bytes, "
+            f"but at most {limit} can be sent at once",
+            path=[CONF_MAX_PACKET_SIZE],
+        )
+
+
 # Fragments from different devices are told apart by the addressing layer's source address, so of two devices sending
 # fragmented packets at once, one still gets through; without addressing, the receiver can drop both
-@FRAGMENTATION.register("basic", schema=BASIC_FRAGMENTATION_SCHEMA)
+@FRAGMENTATION.register(
+    "basic",
+    schema=BASIC_FRAGMENTATION_SCHEMA,
+    packet_size=lambda config, size: config[CONF_MAX_PACKET_SIZE],
+    check_lower=_check_lower,
+)
 async def build_basic_fragmentation(config: ConfigType, lower: MockObj) -> MockObj:
     cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION")
     var = cg.new_Pvariable(config[CONF_ID], lower, config[CONF_MAX_PACKET_SIZE])

esphome/components/packet_interface/layers/hdlc_reliability.py

@@ -6,12 +6,15 @@ from esphome.const import CONF_ID, CONF_TIMEOUT
 from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType
 
-from . import RELIABILITY, PacketInterface, packet_interface_ns
+from . import RELIABILITY, PacketInterface, packet_interface_ns, remaining_packet_size
 
 CONF_ACK_DELAY = "ack_delay"
 CONF_MAX_QUEUE_LENGTH = "max_queue_length"
 CONF_MAX_RETRIES = "max_retries"
 
+# The control byte and the CRC-16
+FRAME_OVERHEAD = 3
+
 HdlcReliability = packet_interface_ns.class_("HdlcReliability", PacketInterface)
 
 
@@ -47,7 +50,13 @@ HDLC_SCHEMA = cv.All(
 )
 
 
-@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA, single_peer=lambda config: True)
+@RELIABILITY.register(
+    "hdlc",
+    schema=HDLC_SCHEMA,
+    single_peer=lambda config: True,
+    packet_size=lambda config, size: remaining_packet_size(size, FRAME_OVERHEAD),
+    send_capacity=lambda config: config[CONF_MAX_QUEUE_LENGTH],
+)
 async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj:
     cg.add_define("USE_PACKET_INTERFACE_HDLC")
     var = cg.new_Pvariable(config[CONF_ID], lower)

esphome/components/packet_interface/layers/xxtea_encryption.py

@@ -8,10 +8,14 @@ from esphome.const import CONF_ID, CONF_KEY
 from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType
 
-from . import ENCRYPTION, PacketInterface, packet_interface_ns
+from . import ENCRYPTION, PacketInterface, packet_interface_ns, remaining_packet_size
 
 CONF_ROLLING_CODE_ENABLE = "rolling_code_enable"
 
+# The check word, and with the rolling code the boot numbers and counter
+CHECK_SIZE = 4
+ROLLING_SIZE = 12
+
 XxteaEncryption = packet_interface_ns.class_("XxteaEncryption", PacketInterface)
 
 XXTEA_SCHEMA = cv.Schema(
@@ -31,6 +35,11 @@ XXTEA_SCHEMA = cv.Schema(
     single_peer=lambda config: config[CONF_ROLLING_CODE_ENABLE],
     needs_session=lambda config: config[CONF_ROLLING_CODE_ENABLE],
     auto_load=lambda config: ["xxtea"],
+    # XXTEA works on whole 32-bit words
+    packet_size=lambda config, size: remaining_packet_size(
+        size & ~3,
+        CHECK_SIZE + (ROLLING_SIZE if config[CONF_ROLLING_CODE_ENABLE] else 0),
+    ),
 )
 async def build_xxtea(config: ConfigType, lower: MockObj) -> MockObj:
     cg.add_define("USE_PACKET_INTERFACE_XXTEA")

tests/components/packet_interface/common.yaml

@@ -29,8 +29,9 @@ packet_interface:
         - id: packet_interface_peer_b
           destination: 0x0003
           reliability:
+            max_queue_length: 5
           fragmentation:
-            max_packet_size: 2048
+            max_packet_size: 1024
           session:
         - id: packet_interface_bus
           destination: broadcast

tests/unit_tests/test_packet_interface.py

@@ -1,4 +1,4 @@
-"""Unit tests for the packet_interface component and its platforms."""
+"""Unit tests for the packet_interface component."""
 
 from typing import Any
 
@@ -72,6 +72,62 @@ def test_reliability_queue_length_limit(setup_core) -> None:
         )
 
 
+def test_reliability_timeout_minimum(setup_core) -> None:
+    """A timeout shorter than a few loop passes cannot work."""
+    uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}})
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}})
+
+
+def test_ack_delay_shorter_than_timeout(setup_core) -> None:
+    """The acknowledgement must not wait longer than the peer's timeout."""
+    reliability = uart_packet_interface.CONFIG_SCHEMA(
+        {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}}
+    )[CONF_RELIABILITY]
+    assert reliability["ack_delay"].total_milliseconds == 100
+    with pytest.raises(cv.Invalid, match="shorter"):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}}
+        )
+
+
+def test_layer_types_registered() -> None:
+    """Importing packet_interface registers the built-in layer types."""
+    assert {kind.key: set(kind.types) for kind in STACK_ORDER} == {
+        CONF_ENCRYPTION: {"xxtea"},
+        CONF_RELIABILITY: {"hdlc"},
+        CONF_FRAGMENTATION: {"basic"},
+        CONF_SESSION: {"basic"},
+    }
+
+
+async def _build(config: dict, lower: Any) -> Any:
+    return lower
+
+
+def test_registered_type_selectable(setup_core) -> None:
+    """A newly registered layer type can be selected with type."""
+    kind = LayerKind("test_kind", "test")
+    kind.register("test", schema=cv.Schema({}))(_build)
+    config = kind.schema()({"type": "test"})
+    assert config["type"] == "test"
+    assert kind.get_type(config).build is _build
+
+
+def test_duplicate_layer_type_rejected(setup_core) -> None:
+    """Registering a type under a name already in use is an error, not a silent replacement."""
+    kind = LayerKind("test_kind", "test")
+    kind.register("test", schema=cv.Schema({}))(_build)
+    with pytest.raises(ValueError, match="already registered"):
+        kind.register("test", schema=cv.Schema({}))(_build)
+
+
+def test_late_registration_rejected(setup_core) -> None:
+    """A type registered after the schemas were built would be missing from them, so it is an error."""
+    with pytest.raises(ValueError, match="after the schemas were built"):
+        RELIABILITY.register("late", schema=cv.Schema({}))(_build)
+
+
 def test_session_defaults(setup_core) -> None:
     """An empty session key selects the basic session, timing out after three keep-alives."""
     session = uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: None})[CONF_SESSION]
@@ -154,7 +210,7 @@ def test_two_device_layers_need_destination(setup_core, layers: dict) -> None:
     )
 
 
-def _peer(peer_id: str, destination: Any, **layers: Any) -> dict:
+def _peer(peer_id: str | ID, destination: Any, **layers: Any) -> dict:
     return {"id": peer_id, "destination": destination, **layers}
 
 
@@ -196,62 +252,6 @@ def test_addressing_peers_invalid(
         )
 
 
-def test_layer_types_registered() -> None:
-    """Importing packet_interface registers the built-in layer types."""
-    assert {kind.key: set(kind.types) for kind in STACK_ORDER} == {
-        CONF_ENCRYPTION: {"xxtea"},
-        CONF_RELIABILITY: {"hdlc"},
-        CONF_FRAGMENTATION: {"basic"},
-        CONF_SESSION: {"basic"},
-    }
-
-
-async def _build(config: dict, lower: Any) -> Any:
-    return lower
-
-
-def test_registered_type_selectable(setup_core) -> None:
-    """A newly registered layer type can be selected with type."""
-    kind = LayerKind("test_kind", "test")
-    kind.register("test", schema=cv.Schema({}))(_build)
-    config = kind.schema()({"type": "test"})
-    assert config["type"] == "test"
-    assert kind.get_type(config).build is _build
-
-
-def test_duplicate_layer_type_rejected(setup_core) -> None:
-    """Registering a type under a name already in use is an error, not a silent replacement."""
-    kind = LayerKind("test_kind", "test")
-    kind.register("test", schema=cv.Schema({}))(_build)
-    with pytest.raises(ValueError, match="already registered"):
-        kind.register("test", schema=cv.Schema({}))(_build)
-
-
-def test_late_registration_rejected(setup_core) -> None:
-    """A type registered after the schemas were built would be missing from them, so it is an error."""
-    with pytest.raises(ValueError, match="after the schemas were built"):
-        RELIABILITY.register("late", schema=cv.Schema({}))(_build)
-
-
-def test_reliability_timeout_minimum(setup_core) -> None:
-    """A timeout shorter than a few loop passes cannot work."""
-    uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}})
-    with pytest.raises(cv.Invalid):
-        uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}})
-
-
-def test_ack_delay_shorter_than_timeout(setup_core) -> None:
-    """The acknowledgement must not wait longer than the peer's timeout."""
-    reliability = uart_packet_interface.CONFIG_SCHEMA(
-        {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}}
-    )[CONF_RELIABILITY]
-    assert reliability["ack_delay"].total_milliseconds == 100
-    with pytest.raises(cv.Invalid, match="shorter"):
-        uart_packet_interface.CONFIG_SCHEMA(
-            {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}}
-        )
-
-
 def test_fragmentation_defaults(setup_core) -> None:
     """An empty fragmentation key joins packets of up to 1024 bytes."""
     fragmentation = uart_packet_interface.CONFIG_SCHEMA({CONF_FRAGMENTATION: None})[
@@ -265,12 +265,71 @@ def test_fragmentation_defaults(setup_core) -> None:
         )
 
 
+@pytest.mark.parametrize(
+    ("layers", "valid"),
+    [
+        # 508 - 3 leaves fragments of 503 bytes: 1024 bytes take 3, 2048 bytes take 5
+        ({CONF_RELIABILITY: None, CONF_FRAGMENTATION: None}, True),
+        (
+            {CONF_RELIABILITY: None, CONF_FRAGMENTATION: {"max_packet_size": 2048}},
+            False,
+        ),
+        (
+            {
+                CONF_RELIABILITY: {CONF_MAX_QUEUE_LENGTH: 5},
+                CONF_FRAGMENTATION: {"max_packet_size": 2048},
+            },
+            True,
+        ),
+        ({CONF_FRAGMENTATION: {"max_packet_size": 16384}}, True),
+    ],
+)
+def test_fragments_fit_reliability_queue(setup_core, layers: dict, valid: bool) -> None:
+    """Every fragment of the largest packet must fit in the reliability queue at once."""
+    if valid:
+        uart_packet_interface.CONFIG_SCHEMA(layers)
+        return
+    with pytest.raises(cv.Invalid, match="5 fragments") as err:
+        uart_packet_interface.CONFIG_SCHEMA(layers)
+    assert err.value.path == [CONF_FRAGMENTATION, CONF_MAX_PACKET_SIZE]
+
+
+@pytest.mark.parametrize(
+    ("link", "match"),
+    [
+        # 256 - 4 for addressing - 16 for the rolling code - 3 for reliability leaves fragments of 231 bytes
+        ({}, "5 fragments of 231 bytes"),
+        # 24 - 4 - 16 - 3 leaves 1 byte, too few for the fragment header
+        ({CONF_MAX_PACKET_SIZE: 24}, "no room"),
+    ],
+)
+def test_fragments_checked_in_peers(setup_core, link: dict, match: str) -> None:
+    """Fragments are checked against the layers of each peer, below the addressing layer."""
+    peer = _peer(
+        "a",
+        2,
+        encryption={"key": "secret", "rolling_code_enable": True},
+        reliability=None,
+        fragmentation=None,
+        session=None,
+    )
+    with pytest.raises(cv.Invalid, match=match) as err:
+        uart_packet_interface.CONFIG_SCHEMA(
+            {
+                CONF_MAX_PACKET_SIZE: 256,
+                **link,
+                CONF_ADDRESSING: {"address": 1, "peers": [peer]},
+            }
+        )
+    assert err.value.path[:4] == [CONF_ADDRESSING, "peers", 0, CONF_FRAGMENTATION]
+
+
 @pytest.mark.parametrize(
     ("config", "loads"),
     [
         ({}, []),
-        ({CONF_RELIABILITY: None, "session": None}, []),
-        ({"session": {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]),
+        ({CONF_RELIABILITY: None, CONF_SESSION: None}, []),
+        ({CONF_SESSION: {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]),
         ({CONF_ENCRYPTION: "secret"}, ["xxtea"]),
         (
             {
@@ -315,7 +374,7 @@ def test_application_needs_peer_id_with_several_peers(setup_core) -> None:
     token = fv.full_config.set(
         {
             "packet_interface": [
-                {"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer("a", 2)]}},
+                {"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer(ID("a"), 2)]}},
                 {"id": ID("single"), CONF_ADDRESSING: {"destination": 2}},
             ]
         }