[packet_interface] Route broadcasts to the broadcast port, and keep a join

A broadcast from a device with its own port went to that port, whose
layers drop it, so a broadcast port could not reach reliable peers. It now
goes to the broadcast port when there is one.

Fragmentation allocates its buffers with RAMAllocator and fails setup when
they cannot be allocated. A fragment from another source no longer drops
the packet being joined.

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

esphome/components/packet_interface/basic_addressing.cpp

@@ -40,16 +40,20 @@ void BasicAddressing::receive_(const PacketBuffer &data, const PacketMetaData &m
   uint16_t source = encode_uint16(packet[3], packet[2]);
   if (source == this->address_ || (destination != this->address_ && destination != BROADCAST_ADDRESS))
     return;
-  // A device with a port of its own is heard only there; the broadcast port hears the rest
-  AddressingPort *target = nullptr;
+  AddressingPort *source_port = nullptr;
+  AddressingPort *broadcast_port = nullptr;
   for (auto *port : this->ports_) {
     if (port->destination_ == source) {
-      target = port;
-      break;
+      source_port = port;
+    } else if (port->destination_ == BROADCAST_ADDRESS) {
+      broadcast_port = port;
     }
-    if (port->destination_ == BROADCAST_ADDRESS)
-      target = port;
   }
+  // A broadcast goes to the broadcast port, even from a device with a port of its own, as that port's layers expect
+  // only its point-to-point packets; anything else goes to the source's port, or to the broadcast port if it has none
+  AddressingPort *target = source_port;
+  if (broadcast_port != nullptr && (destination == BROADCAST_ADDRESS || source_port == nullptr))
+    target = broadcast_port;
   if (target == nullptr) {
     ESP_LOGV(TAG, "%s: Packet from 0x%04X dropped", this->link_->get_name(), source);
     return;

esphome/components/packet_interface/basic_addressing.h

@@ -19,7 +19,8 @@ class AddressingPort;
  * Above this layer are one or more ports, each the bottom of its own stack of layers. A port with a destination
  * address is a point-to-point link to that device: it sends to it and receives only its packets, so the reliability
  * and session layers and the rolling code, which work between two devices only, can be used above it. A port with
- * the broadcast destination sends to every device, and receives packets from devices that have no port of their own.
+ * the broadcast destination sends to every device, and receives every broadcast, as well as packets from devices
+ * that have no port of their own. Without a broadcast port, a device's broadcasts go to its own port.
  */
 class BasicAddressing : public Component {
  public:

esphome/components/packet_interface/basic_fragmentation.cpp

@@ -33,10 +33,16 @@ void BasicFragmentation::setup() {
     this->mark_failed(LOG_STR("max_packet_size needs more than 128 fragments"));
     return;
   }
-  // Checked now, while the layer below holds nothing
-  this->queue_too_short_ = this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity();
-  this->tx_buffer_ = std::make_unique<uint8_t[]>(this->fragment_size_());
-  this->rx_buffer_ = std::make_unique<uint8_t[]>(this->max_packet_size_);
+  // Checked now, while the layer below holds nothing; a failed layer below reports its own failure instead
+  this->queue_too_short_ =
+      !this->lower_->is_failed() && this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity();
+  RAMAllocator<uint8_t> allocator;
+  this->tx_buffer_ = allocator.allocate(this->fragment_size_() + this->max_packet_size_);
+  if (this->tx_buffer_ == nullptr) {
+    this->mark_failed(LOG_STR("Not enough memory for the packet buffers"));
+    return;
+  }
+  this->rx_buffer_ = this->tx_buffer_ + this->fragment_size_();
   // Numbering from a random start makes it unlikely that a packet joins fragments from before a restart
   this->tx_number_ = random_uint32();
   this->lower_->add_packet_interface_listener(
@@ -58,7 +64,7 @@ void BasicFragmentation::dump_config() {
 }
 
 bool BasicFragmentation::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
-  if (this->is_failed())
+  if (this->tx_buffer_ == nullptr)
     return false;
   size_t size = data.size();
   if (size > this->max_packet_size_) {
@@ -77,9 +83,9 @@ bool BasicFragmentation::send_to_interface(const PacketBuffer &data, const Packe
   }
   size_t fragment_size = this->fragment_size_();
   for (size_t index = 0; index != fragments; index++) {
-    size_t length = data.copy_to(this->tx_buffer_.get(), fragment_size, index * fragment_size);
+    size_t length = data.copy_to(this->tx_buffer_, fragment_size, index * fragment_size);
     header[1] = index | (index + 1 == fragments ? LAST_FRAGMENT : 0);
-    PacketBuffer body(this->tx_buffer_.get(), length);
+    PacketBuffer body(this->tx_buffer_, length);
     if (!this->lower_->send_to_interface(PacketBuffer(header, HEADER_SIZE, &body, nullptr, 0), meta_data)) {
       // The peer drops the part already sent when the next packet starts
       ESP_LOGV(TAG, "%s: Fragment %zu of %zu not accepted", this->name_, index, fragments);
@@ -103,26 +109,33 @@ void BasicFragmentation::receive_(const PacketBuffer &data, const PacketMetaData
   bool last = (fragment[1] & LAST_FRAGMENT) != 0;
   const uint8_t *payload = fragment.data() + HEADER_SIZE;
   size_t length = fragment.size() - HEADER_SIZE;
+  if (index == 0 && last) {
+    // A packet in one fragment leaves a packet being joined from another device intact
+    if (this->rx_joining_ && meta_data.source_address == this->rx_source_) {
+      ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_);
+      this->rx_joining_ = false;
+    }
+    if (length > this->max_packet_size_) {
+      ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_);
+      return;
+    }
+    this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data);
+    return;
+  }
   if (index == 0) {
     if (this->rx_joining_) {
       ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_);
     }
-    this->rx_joining_ = false;
-    if (last) {
-      if (length > this->max_packet_size_) {
-        ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_);
-        return;
-      }
-      this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data);
-      return;
-    }
     this->rx_joining_ = true;
     this->rx_number_ = number;
     this->rx_source_ = meta_data.source_address;
     this->rx_next_index_ = 0;
     this->rx_length_ = 0;
-  } else if (!this->rx_joining_ || number != this->rx_number_ || meta_data.source_address != this->rx_source_ ||
-             index != this->rx_next_index_) {
+  } else if (this->rx_joining_ && meta_data.source_address != this->rx_source_) {
+    // Only this fragment is lost, so of two devices sending at once, the one that started last still gets through
+    ESP_LOGV(TAG, "%s: Fragment %u from 0x%04X dropped", this->name_, index, meta_data.source_address);
+    return;
+  } else if (!this->rx_joining_ || number != this->rx_number_ || index != this->rx_next_index_) {
     ESP_LOGV(TAG, "%s: Fragment %u of packet %u dropped", this->name_, index, number);
     this->rx_joining_ = false;
     return;
@@ -132,12 +145,12 @@ void BasicFragmentation::receive_(const PacketBuffer &data, const PacketMetaData
     this->rx_joining_ = false;
     return;
   }
-  memcpy(this->rx_buffer_.get() + this->rx_length_, payload, length);
+  memcpy(this->rx_buffer_ + this->rx_length_, payload, length);
   this->rx_length_ += length;
   this->rx_next_index_++;
   if (last) {
     this->rx_joining_ = false;
-    this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_.get(), this->rx_length_), meta_data);
+    this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_, this->rx_length_), meta_data);
   }
 }
 

esphome/components/packet_interface/basic_fragmentation.h

@@ -2,7 +2,6 @@
 
 #include "esphome/core/defines.h"
 #ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION
-#include <memory>
 
 #include "packet_interface.h"
 
@@ -18,8 +17,9 @@ namespace esphome::packet_interface {
  *
  * A packet is started only if the layer below can take all of its fragments now (get_send_capacity()), so a full
  * reliability queue does not leave the peer with part of a packet. The receiver joins one packet at a time, and drops
- * it when a fragment is missing, or one arrives with another packet number or from another source address. Above a
- * reliability layer, fragments arrive in order and none are missing.
+ * it when a fragment is missing or arrives with another packet number, or when the same source, or another one,
+ * starts a new packet of several fragments. Other fragments from another source address are dropped, leaving the
+ * packet being joined intact. Above a reliability layer, fragments arrive in order and none are missing.
  */
 class BasicFragmentation : public PacketInterface {
  public:
@@ -43,9 +43,10 @@ class BasicFragmentation : public PacketInterface {
   size_t fragments_for_(size_t size) const;
 
   PacketInterface *lower_;
-  // One fragment being sent, and the packet being joined; both allocated in setup()
-  std::unique_ptr<uint8_t[]> tx_buffer_;
-  std::unique_ptr<uint8_t[]> rx_buffer_;
+  // One fragment being sent, and the packet being joined; one block allocated in setup(), and tx_buffer_ stays null
+  // if that fails
+  uint8_t *tx_buffer_{nullptr};
+  uint8_t *rx_buffer_{nullptr};
   size_t rx_length_{0};
   uint16_t rx_source_{0};
   uint8_t tx_number_{0};

esphome/components/packet_interface/layers/basic_fragmentation.py

@@ -21,8 +21,8 @@ BASIC_FRAGMENTATION_SCHEMA = cv.Schema(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
-# Fragments from different devices are told apart by the addressing layer's source address, so several devices can
-# share a channel; without addressing, two devices sending fragmented packets at once can make the receiver drop both
+# 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)
 async def build_basic_fragmentation(config: ConfigType, lower: MockObj) -> MockObj:
     cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION")

tests/components/packet_interface/basic_addressing.cpp

@@ -143,6 +143,24 @@ TEST(BasicAddressingTest, PeersHaveSeparateStacks) {
   EXPECT_TRUE(d.port().received.empty());
 }
 
+TEST(BasicAddressingTest, BroadcastFromPeerReachesBroadcastPort) {
+  Device hub(1, {2, BROADCAST}), b(2, {1, BROADCAST});
+  hub.port(1).send({0x10});  // a state every device needs
+  hub.port(0).send({0x11});  // point to point
+  hub.deliver({&b});
+  EXPECT_EQ(b.port(0).received, std::vector<Packet>{{0x11}});
+  EXPECT_EQ(b.port(1).received, std::vector<Packet>{{0x10}});
+  EXPECT_EQ(b.port(1).sources, std::vector<uint16_t>{1});
+}
+
+TEST(BasicAddressingTest, WithoutBroadcastPortPeerBroadcastReachesItsPort) {
+  Device hub(1, {2, 3}), b(2, BROADCAST);
+  b.port().send({0x20});
+  b.deliver({&hub});
+  EXPECT_EQ(hub.port(0).received, std::vector<Packet>{{0x20}});
+  EXPECT_TRUE(hub.port(1).received.empty());
+}
+
 TEST(BasicAddressingTest, WithoutBroadcastPortUnknownDevicesDropped) {
   Device hub(1, {2, 3}), d(4);
   d.port().send({0x04});

tests/components/packet_interface/basic_fragmentation.cpp

@@ -161,8 +161,21 @@ TEST_F(BasicFragmentationTest, InterleavedSourcesNotMixed) {
   this->link_b_.inject(from_c[0], 2);
   this->link_b_.inject(from_a[1], 1);
   this->link_b_.inject(from_c[1], 2);
-  for (auto &packet : b.received)
-    EXPECT_EQ(packet, make_packet(20));  // nothing mixed from the two
+  // Nothing is mixed from the two, and the packet started last gets through
+  EXPECT_EQ(b.received, std::vector<Packet>{make_packet(20)});
+}
+
+TEST_F(BasicFragmentationTest, OneFragmentPacketFromAnotherSourceKeepsJoin) {
+  Peer a(&this->link_a_), c(&this->link_a_), b(&this->link_b_);
+  a.send(make_packet(20));
+  auto from_a = std::move(this->link_a_.sent);
+  this->link_a_.sent.clear();
+  c.send(make_packet(5));
+  auto from_c = std::move(this->link_a_.sent);
+  this->link_b_.inject(from_a[0], 1);
+  this->link_b_.inject(from_c[0], 2);
+  this->link_b_.inject(from_a[1], 1);
+  EXPECT_EQ(b.received, (std::vector<Packet>{make_packet(5), make_packet(20)}));
 }
 
 TEST_F(BasicFragmentationTest, OverReliabilityWithLosses) {