mirror of
https://github.com/esphome/esphome.git
synced 2026-10-21 13:34:12 +00:00
[entity_transport] Ask again for a dictionary with a lost part, and warn on dropped packets
A dictionary packet now carries its part index and a last-part flag. A dictionary counts as received, and is not asked for again, only once its last part follows every other part in order. Until then, data under it sends a request even when some remote entities are already known. A refused request no longer starts the request interval. A short or malformed packet, or one with another format version, is logged as a warning at most once a minute. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
75c91fb6e5
commit
88554b0354
@@ -13,6 +13,7 @@ using packet_interface::PacketMetaData;
|
||||
ESPHOME_LOG_TAG(TAG, "entity_transport");
|
||||
|
||||
// The packet type byte holds the format version in its high nibble
|
||||
static constexpr uint8_t FORMAT_VERSION = 1;
|
||||
static constexpr uint8_t PACKET_DICTIONARY = 0x11;
|
||||
static constexpr uint8_t PACKET_DATA = 0x12;
|
||||
static constexpr uint8_t PACKET_REQUEST = 0x13;
|
||||
@@ -22,6 +23,10 @@ static constexpr uint8_t TYPE_BINARY_SENSOR = 2;
|
||||
static constexpr uint8_t MAX_TYPE = 31;
|
||||
// Packet type and dictionary id
|
||||
static constexpr size_t HEADER_SIZE = 5;
|
||||
// A dictionary packet's part byte follows the header: the part's index in bits 0-6, and on the last part, bit 7
|
||||
static constexpr uint8_t PART_INDEX_MASK = 0x7F;
|
||||
static constexpr uint8_t PART_LAST = 0x80;
|
||||
static constexpr uint8_t PART_NONE = 0xFF; // no dictionary is being received in order
|
||||
// A data record starts with a 16-bit tag: the entity's number in bits 0-7, its type in bits 8-12, the payload form in
|
||||
// bits 13-14, and with PAYLOAD_NONE, the state in bit 15. Any record can be skipped without knowing its type.
|
||||
static constexpr size_t TAG_SIZE = 2;
|
||||
@@ -42,6 +47,7 @@ static constexpr uint32_t REQUEST_INTERVAL_MS = 5000;
|
||||
static constexpr uint32_t RETRY_DELAY_MS = 1000;
|
||||
static constexpr uint32_t RETRY_TIMER_ID = 0;
|
||||
static constexpr uint32_t TIMEOUT_CHECK_INTERVAL_MS = 1000;
|
||||
static constexpr uint32_t WARNING_INTERVAL_MS = 60000;
|
||||
|
||||
static void put_uint32(uint8_t *buf, uint32_t value) {
|
||||
for (size_t i = 0; i != 4; i++)
|
||||
@@ -105,8 +111,8 @@ static bool data_well_formed(std::span<const uint8_t> packet) {
|
||||
|
||||
// Checks the device name and every section of a dictionary packet before anything is learnt from it
|
||||
static bool dictionary_well_formed(std::span<const uint8_t> packet) {
|
||||
size_t pos = HEADER_SIZE;
|
||||
if (pos == packet.size())
|
||||
size_t pos = HEADER_SIZE + 1;
|
||||
if (pos >= packet.size())
|
||||
return false;
|
||||
pos += 1 + packet[pos];
|
||||
if (pos > packet.size())
|
||||
@@ -141,7 +147,7 @@ void EntityTransport::setup() {
|
||||
longest_id = std::max(longest_id, strlen(entity.id));
|
||||
// A dictionary packet must hold the device name and any one id, which also leaves room for a data record
|
||||
if (name.size() > MAX_STRING_LENGTH || longest_id > MAX_STRING_LENGTH ||
|
||||
HEADER_SIZE + 1 + name.size() + SECTION_HEADER_SIZE + 1 + longest_id > this->max_packet_size_) {
|
||||
HEADER_SIZE + 2 + name.size() + SECTION_HEADER_SIZE + 1 + longest_id > this->max_packet_size_) {
|
||||
this->mark_failed(LOG_STR("max_packet_size too small for the device name and entity ids"));
|
||||
return;
|
||||
}
|
||||
@@ -236,6 +242,7 @@ void EntityTransport::dump_config() {
|
||||
void EntityTransport::send_all_() {
|
||||
this->dictionary_pending_ = true;
|
||||
this->dictionary_next_ = 0;
|
||||
this->dictionary_part_ = 0;
|
||||
for (auto &entity : this->entities_)
|
||||
entity.updated = true;
|
||||
this->enable_loop();
|
||||
@@ -263,6 +270,7 @@ bool EntityTransport::send_dictionary_() {
|
||||
uint8_t *buf = this->buffer_;
|
||||
const auto &name = App.get_name();
|
||||
size_t header = this->write_header_(PACKET_DICTIONARY);
|
||||
size_t part_offset = header++;
|
||||
buf[header++] = name.size();
|
||||
memcpy(buf + header, name.c_str(), name.size());
|
||||
header += name.size();
|
||||
@@ -276,9 +284,11 @@ bool EntityTransport::send_dictionary_() {
|
||||
size_t id_length = strlen(entity.id);
|
||||
bool new_section = type != section_type;
|
||||
if (length + (new_section ? SECTION_HEADER_SIZE : 0) + 1 + id_length > this->max_packet_size_) {
|
||||
buf[part_offset] = this->dictionary_part_ & PART_INDEX_MASK;
|
||||
if (!this->interface_->send_to_interface(PacketBuffer(buf, length), {}))
|
||||
return false;
|
||||
this->dictionary_next_ = i;
|
||||
this->dictionary_part_++;
|
||||
length = header;
|
||||
new_section = true;
|
||||
}
|
||||
@@ -294,10 +304,14 @@ bool EntityTransport::send_dictionary_() {
|
||||
memcpy(buf + length, entity.id, id_length);
|
||||
length += id_length;
|
||||
}
|
||||
if (length != header && !this->interface_->send_to_interface(PacketBuffer(buf, length), {}))
|
||||
return false;
|
||||
if (length != header) {
|
||||
buf[part_offset] = (this->dictionary_part_ & PART_INDEX_MASK) | PART_LAST;
|
||||
if (!this->interface_->send_to_interface(PacketBuffer(buf, length), {}))
|
||||
return false;
|
||||
}
|
||||
this->dictionary_pending_ = false;
|
||||
this->dictionary_next_ = 0;
|
||||
this->dictionary_part_ = 0;
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -377,8 +391,19 @@ void EntityTransport::receive_(const PacketBuffer &data) {
|
||||
return;
|
||||
}
|
||||
auto packet = data.as_span();
|
||||
if (packet.size() < HEADER_SIZE)
|
||||
if (packet.size() < HEADER_SIZE) {
|
||||
if (this->warning_due_()) {
|
||||
ESP_LOGW(TAG, "Packet of %zu bytes is too short", packet.size());
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (packet[0] >> 4 != FORMAT_VERSION) {
|
||||
if (this->warning_due_()) {
|
||||
ESP_LOGW(TAG, "Packet format version %u not supported; the sender may run another ESPHome version",
|
||||
packet[0] >> 4);
|
||||
}
|
||||
return;
|
||||
}
|
||||
uint32_t dictionary_id = encode_uint32(packet[4], packet[3], packet[2], packet[1]);
|
||||
if (dictionary_id == this->dictionary_id_) {
|
||||
// Either a request for our dictionary, or our own packet heard on a broadcast channel
|
||||
@@ -399,11 +424,19 @@ void EntityTransport::receive_(const PacketBuffer &data) {
|
||||
|
||||
void EntityTransport::receive_dictionary_(std::span<const uint8_t> packet, uint32_t dictionary_id) {
|
||||
if (!dictionary_well_formed(packet)) {
|
||||
ESP_LOGV(TAG, "Dictionary packet of %zu bytes is malformed", packet.size());
|
||||
if (this->warning_due_()) {
|
||||
ESP_LOGW(TAG, "Dictionary packet of %zu bytes is malformed", packet.size());
|
||||
}
|
||||
return;
|
||||
}
|
||||
size_t pos = HEADER_SIZE;
|
||||
if (!this->dictionary_seen_(dictionary_id)) {
|
||||
uint8_t part = packet[pos++];
|
||||
uint8_t index = part & PART_INDEX_MASK;
|
||||
// The dictionary is complete, and need not be asked for again, once its last part follows all the others in order
|
||||
bool in_order = index == 0 || (this->rx_dictionary_id_ == dictionary_id && index == this->rx_next_part_);
|
||||
this->rx_dictionary_id_ = dictionary_id;
|
||||
this->rx_next_part_ = in_order ? (index + 1) & PART_INDEX_MASK : PART_NONE;
|
||||
if (in_order && (part & PART_LAST) != 0 && !this->dictionary_seen_(dictionary_id)) {
|
||||
this->seen_dictionaries_[this->seen_next_] = dictionary_id;
|
||||
this->seen_next_ = (this->seen_next_ + 1) % this->seen_dictionaries_.size();
|
||||
}
|
||||
@@ -470,15 +503,16 @@ void EntityTransport::receive_data_(std::span<const uint8_t> packet, uint32_t di
|
||||
has_remotes = true;
|
||||
}
|
||||
#endif
|
||||
if (!known) {
|
||||
// A dictionary not yet received may name our entities, or be a provider's new one; one already received named
|
||||
// none of them, so asking again would only make that provider resend everything
|
||||
if (has_remotes && !this->dictionary_seen_(dictionary_id))
|
||||
this->request_dictionary_(dictionary_id);
|
||||
// A dictionary not yet received in full may name our entities, or be a provider's new one; one received in full
|
||||
// named all of them it could, so asking again would only make that provider resend everything
|
||||
if (has_remotes && !this->dictionary_seen_(dictionary_id))
|
||||
this->request_dictionary_(dictionary_id);
|
||||
if (!known)
|
||||
return;
|
||||
}
|
||||
if (!data_well_formed(packet)) {
|
||||
ESP_LOGV(TAG, "Data packet of %zu bytes is malformed", packet.size());
|
||||
if (this->warning_due_()) {
|
||||
ESP_LOGW(TAG, "Data packet of %zu bytes is malformed", packet.size());
|
||||
}
|
||||
return;
|
||||
}
|
||||
[[maybe_unused]] uint32_t now = App.get_loop_component_start_time();
|
||||
@@ -522,12 +556,26 @@ void EntityTransport::request_dictionary_(uint32_t dictionary_id) {
|
||||
uint32_t now = App.get_loop_component_start_time();
|
||||
if (this->request_sent_ && now - this->last_request_ms_ < REQUEST_INTERVAL_MS)
|
||||
return;
|
||||
this->request_sent_ = true;
|
||||
this->last_request_ms_ = now;
|
||||
ESP_LOGD(TAG, "Requesting dictionary %08" PRIX32, dictionary_id);
|
||||
uint8_t packet[HEADER_SIZE] = {PACKET_REQUEST};
|
||||
put_uint32(packet + 1, dictionary_id);
|
||||
this->interface_->send_to_interface(PacketBuffer(packet, sizeof(packet)), {});
|
||||
// A refused request is tried again with the next data packet under that dictionary id
|
||||
if (!this->interface_->send_to_interface(PacketBuffer(packet, sizeof(packet)), {})) {
|
||||
ESP_LOGV(TAG, "Request for dictionary %08" PRIX32 " not accepted", dictionary_id);
|
||||
return;
|
||||
}
|
||||
this->request_sent_ = true;
|
||||
this->last_request_ms_ = now;
|
||||
ESP_LOGD(TAG, "Requested dictionary %08" PRIX32, dictionary_id);
|
||||
}
|
||||
|
||||
// Dropped packets point to a device with other firmware or a fault, and may arrive steadily, so warn only now and then
|
||||
bool EntityTransport::warning_due_() {
|
||||
uint32_t now = App.get_loop_component_start_time();
|
||||
if (this->warned_ && now - this->last_warning_ms_ < WARNING_INTERVAL_MS)
|
||||
return false;
|
||||
this->warned_ = true;
|
||||
this->last_warning_ms_ = now;
|
||||
return true;
|
||||
}
|
||||
|
||||
// A newly connected or restarted peer needs the dictionary and every state; while it is lost, its entities have no
|
||||
|
||||
@@ -29,6 +29,8 @@ namespace esphome::entity_transport {
|
||||
* Dictionary packet:
|
||||
* - type: 1 byte, PACKET_DICTIONARY;
|
||||
* - dictionary id: 4 bytes;
|
||||
* - part: 1 byte, the packet's index in the dictionary in bits 0-6, counting from 0 and wrapping, and bit 7 set on
|
||||
* the last packet;
|
||||
* - sender's device name: length (1 byte), then the name;
|
||||
* - one or more sections, each: entity type (1 byte), first number (1 byte), count (1 byte), then for each entity
|
||||
* its id: length (1 byte), then the id.
|
||||
@@ -43,8 +45,8 @@ namespace esphome::entity_transport {
|
||||
*
|
||||
* Every packet is checked in full before anything in it is used, so a damaged or forged packet changes no state.
|
||||
*
|
||||
* Dictionary request packet, sent by a device that receives data with a dictionary id it has not received a dictionary
|
||||
* for:
|
||||
* Dictionary request packet, sent by a device that receives data with a dictionary id it has not received in full,
|
||||
* with every part in order:
|
||||
* - type: 1 byte, PACKET_REQUEST;
|
||||
* - the dictionary id wanted: 4 bytes.
|
||||
*
|
||||
@@ -123,6 +125,7 @@ class EntityTransport : public Component {
|
||||
size_t id_length, uint32_t dictionary_id);
|
||||
void request_dictionary_(uint32_t dictionary_id);
|
||||
bool dictionary_seen_(uint32_t dictionary_id) const;
|
||||
bool warning_due_();
|
||||
void on_session_(bool connected);
|
||||
void clear_remote_states_(bool timed_out_only);
|
||||
|
||||
@@ -142,14 +145,20 @@ class EntityTransport : public Component {
|
||||
uint32_t remote_timeout_ms_{0};
|
||||
uint32_t dictionary_id_{0};
|
||||
uint32_t last_request_ms_{0};
|
||||
// The last dictionary ids received, so a provider whose dictionary names none of our entities is not asked again
|
||||
uint32_t last_warning_ms_{0};
|
||||
uint32_t rx_dictionary_id_{0}; // the dictionary whose parts are arriving
|
||||
// The last dictionary ids received in full, so a provider whose dictionary names none of our entities is not asked
|
||||
// again
|
||||
std::array<uint32_t, 8> seen_dictionaries_{};
|
||||
uint8_t seen_next_{0};
|
||||
uint8_t rx_next_part_{0}; // the index of its next part, or PART_NONE if one was missed
|
||||
uint8_t dictionary_part_{0}; // the index of the next dictionary packet to send
|
||||
uint8_t sensor_count_{0};
|
||||
uint8_t binary_sensor_count_{0};
|
||||
size_t dictionary_next_{0}; // the first entity of the next dictionary packet to send
|
||||
bool dictionary_pending_{true};
|
||||
bool request_sent_{false};
|
||||
bool warned_{false};
|
||||
};
|
||||
|
||||
} // namespace esphome::entity_transport
|
||||
|
||||
@@ -66,6 +66,15 @@ static Packet header(uint8_t packet_type, uint32_t id) {
|
||||
static_cast<uint8_t>(id >> 24)};
|
||||
}
|
||||
|
||||
// The start of a dictionary packet from this device's name: the header, the part byte, then the name
|
||||
static Packet dictionary_start(uint32_t id, uint8_t part = 0x80) {
|
||||
Packet packet = header(PACKET_DICTIONARY, id);
|
||||
packet.push_back(part);
|
||||
packet.push_back(strlen(DEVICE_NAME));
|
||||
packet.insert(packet.end(), DEVICE_NAME, DEVICE_NAME + strlen(DEVICE_NAME));
|
||||
return packet;
|
||||
}
|
||||
|
||||
class EntityTransportTest : public ::testing::Test {
|
||||
protected:
|
||||
void SetUp() override {
|
||||
@@ -110,9 +119,7 @@ TEST_F(EntityTransportTest, SendsDictionaryThenStates) {
|
||||
auto sender = this->make_sender_();
|
||||
sender->loop();
|
||||
uint32_t id = sender->get_dictionary_id();
|
||||
Packet dictionary = header(PACKET_DICTIONARY, id);
|
||||
dictionary.push_back(strlen(DEVICE_NAME));
|
||||
dictionary.insert(dictionary.end(), DEVICE_NAME, DEVICE_NAME + strlen(DEVICE_NAME));
|
||||
Packet dictionary = dictionary_start(id);
|
||||
dictionary.insert(dictionary.end(), {1, 0, 1, 4, 't', 'e', 'm', 'p', 2, 0, 1, 4, 'd', 'o', 'o', 'r'});
|
||||
Packet data = header(PACKET_DATA, id);
|
||||
// Tags: sensor 0 with a 4 byte state, then binary sensor 0 with its state in the top bit
|
||||
@@ -155,7 +162,7 @@ TEST_F(EntityTransportTest, StatesReachRemoteEntities) {
|
||||
}
|
||||
|
||||
TEST_F(EntityTransportTest, SplitsAcrossPackets) {
|
||||
MockInterface link(5 + 1 + strlen(DEVICE_NAME) + 3 + 5);
|
||||
MockInterface link(5 + 2 + strlen(DEVICE_NAME) + 3 + 5);
|
||||
this->temp_.publish_state(1.0f);
|
||||
this->door_.publish_state(true);
|
||||
auto sender = this->make_sender_(&link);
|
||||
@@ -219,6 +226,48 @@ TEST_F(EntityTransportTest, ChangedDictionaryRequested) {
|
||||
EXPECT_EQ(this->remote_temp_.get_state(), 6.0f);
|
||||
}
|
||||
|
||||
TEST_F(EntityTransportTest, LostDictionaryPartRequested) {
|
||||
// One entity in each dictionary packet
|
||||
MockInterface link(5 + 2 + strlen(DEVICE_NAME) + 3 + 5);
|
||||
this->temp_.publish_state(2.0f);
|
||||
this->door_.publish_state(true);
|
||||
auto sender = this->make_sender_(&link);
|
||||
auto receiver = this->make_receiver_();
|
||||
sender->loop();
|
||||
ASSERT_EQ(link.sent.size(), 3);
|
||||
EXPECT_EQ(link.sent[0][5], 0x00);
|
||||
EXPECT_EQ(link.sent[1][5], 0x81);
|
||||
link.sent.erase(link.sent.begin() + 1); // the part naming the binary sensor is lost
|
||||
link.deliver_to(this->link_b_);
|
||||
EXPECT_EQ(this->remote_temp_.get_state(), 2.0f);
|
||||
EXPECT_FALSE(this->remote_door_.has_state());
|
||||
// The sensor is known, but the dictionary is not complete, so it is asked for
|
||||
ASSERT_EQ(this->link_b_.sent.size(), 1);
|
||||
EXPECT_EQ(this->link_b_.sent[0], header(PACKET_REQUEST, sender->get_dictionary_id()));
|
||||
this->link_b_.deliver_to(link);
|
||||
sender->loop();
|
||||
link.deliver_to(this->link_b_);
|
||||
EXPECT_TRUE(this->remote_door_.state);
|
||||
// Now complete, it is not asked for again
|
||||
{ LoopBlockingGuard guard{nullptr, nullptr, 10000}; }
|
||||
this->door_.publish_state(false);
|
||||
sender->loop();
|
||||
link.deliver_to(this->link_b_);
|
||||
EXPECT_FALSE(this->remote_door_.state);
|
||||
EXPECT_TRUE(this->link_b_.sent.empty());
|
||||
}
|
||||
|
||||
TEST_F(EntityTransportTest, RefusedRequestSentAgain) {
|
||||
auto receiver = this->make_receiver_();
|
||||
Packet data = header(PACKET_DATA, 0x12345678);
|
||||
this->link_b_.accept = false;
|
||||
this->link_b_.inject(data);
|
||||
this->link_b_.accept = true;
|
||||
this->link_b_.inject(data);
|
||||
ASSERT_EQ(this->link_b_.sent.size(), 1);
|
||||
EXPECT_EQ(this->link_b_.sent[0], header(PACKET_REQUEST, 0x12345678));
|
||||
}
|
||||
|
||||
TEST_F(EntityTransportTest, RequestsLimited) {
|
||||
auto receiver = this->make_receiver_();
|
||||
Packet data = header(PACKET_DATA, 0x12345678);
|
||||
@@ -307,9 +356,7 @@ TEST_F(EntityTransportRecordTest, UnknownTypesSkipped) {
|
||||
TEST_F(EntityTransportTest, PartlyValidDictionaryNotLearnt) {
|
||||
auto receiver = this->make_receiver_();
|
||||
static constexpr uint32_t OTHER_ID = 0x0BADF00D;
|
||||
Packet dictionary = header(PACKET_DICTIONARY, OTHER_ID);
|
||||
dictionary.push_back(strlen(DEVICE_NAME));
|
||||
dictionary.insert(dictionary.end(), DEVICE_NAME, DEVICE_NAME + strlen(DEVICE_NAME));
|
||||
Packet dictionary = dictionary_start(OTHER_ID);
|
||||
// A good section naming the sensor, then sections cut short, of type 0, and of more than 255 numbers
|
||||
dictionary.insert(dictionary.end(), {1, 0, 1, 4, 't', 'e', 'm', 'p'});
|
||||
for (Packet bad : {Packet{2, 0, 2, 4, 'd', 'o', 'o', 'r'}, Packet{0, 0, 1, 1, 'x'}, Packet{2, 255, 2, 1, 'a', 1, 'b'},
|
||||
@@ -357,7 +404,7 @@ TEST_F(EntityTransportTest, SessionConnectSendsAllAndLossClearsRemote) {
|
||||
|
||||
TEST_F(EntityTransportTest, RefusedDictionaryPartResumes) {
|
||||
// One entity in each dictionary packet
|
||||
MockInterface link(5 + 1 + strlen(DEVICE_NAME) + 3 + 5);
|
||||
MockInterface link(5 + 2 + strlen(DEVICE_NAME) + 3 + 5);
|
||||
this->temp_.publish_state(4.0f);
|
||||
this->door_.publish_state(true);
|
||||
auto sender = this->make_sender_(&link);
|
||||
@@ -397,9 +444,7 @@ TEST_F(EntityTransportTest, RemoteStatesTimeOut) {
|
||||
TEST_F(EntityTransportTest, IdMatchingOnlyUpToNulNotLearnt) {
|
||||
auto receiver = this->make_receiver_();
|
||||
static constexpr uint32_t OTHER_ID = 0x0BADF00D;
|
||||
Packet dictionary = header(PACKET_DICTIONARY, OTHER_ID);
|
||||
dictionary.push_back(strlen(DEVICE_NAME));
|
||||
dictionary.insert(dictionary.end(), DEVICE_NAME, DEVICE_NAME + strlen(DEVICE_NAME));
|
||||
Packet dictionary = dictionary_start(OTHER_ID);
|
||||
// The id "temp" followed by a NUL and more bytes, nine in all
|
||||
dictionary.insert(dictionary.end(), {1, 0, 1, 9, 't', 'e', 'm', 'p', 0, 'a', 'b', 'c', 'd'});
|
||||
this->link_b_.inject(dictionary);
|
||||
@@ -411,7 +456,7 @@ TEST_F(EntityTransportTest, IdMatchingOnlyUpToNulNotLearnt) {
|
||||
|
||||
TEST_F(EntityTransportTest, PacketTooSmallForIdsFailsSetup) {
|
||||
// One byte short of a dictionary packet holding the device name and a four character id
|
||||
MockInterface small_link(5 + 1 + strlen(DEVICE_NAME) + 3 + 1 + 4 - 1);
|
||||
MockInterface small_link(5 + 2 + strlen(DEVICE_NAME) + 3 + 1 + 4 - 1);
|
||||
auto sender = this->make_sender_(&small_link);
|
||||
EXPECT_TRUE(sender->is_failed());
|
||||
EXPECT_FALSE(this->make_sender_()->is_failed());
|
||||
|
||||
Reference in New Issue
Block a user