Skip to content

Log unique ID conflicts #683

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
TheJulianJES wants to merge 2 commits into
base: dev
Choose a base branch
from
Draft

Log unique ID conflicts #683

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d103a3c
WIP: Log duplicate unique IDs
TheJulianJES Feb 27, 2026
0aa985b
WIP: Clean up a bit
TheJulianJES Feb 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions zha/zigbee/device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -976,6 +976,51 @@ def _discover_new_entities(self) -> None:
entity.on_add()
self._pending_entities.append(entity)

@staticmethod
def _entity_class_path(entity: BaseEntity) -> tuple[str, str]:
"""Return module and class name for an entity."""
return (entity.__class__.__module__, entity.__class__.__name__)

@staticmethod
def _entity_endpoint_id(entity: BaseEntity) -> int | None:
"""Return endpoint id if this is a platform entity."""
if not isinstance(entity, PlatformEntity):
return None
return entity.endpoint.id

@staticmethod
def _entity_cluster_handler_names(entity: BaseEntity) -> list[str]:
"""Return cluster handler names if this is a platform entity."""
if not isinstance(entity, PlatformEntity):
return []
return sorted(entity.cluster_handlers.keys())

def _log_entity_unique_id_conflict(
self,
key: tuple[Platform, str],
existing_entity: BaseEntity,
duplicate_entity: BaseEntity,

@zigpy-review-bot zigpy-review-bot May 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

existing_entity/duplicate_entity typed as BaseEntity but the call site (line 1060) only ever passes PlatformEntity — new_entities is dict[tuple[Platform, str], PlatformEntity] and entity reached this point via entity.PLATFORM / entity.unique_id. The isinstance(entity, PlatformEntity) guards in _entity_endpoint_id / _entity_cluster_handler_names are dead branches.

Either narrow these to PlatformEntity and drop the guards (cleaner), or leave the guards in and add a comment explaining the future use case. Current shape is type-annotation overhead with no payoff.

) -> None:
"""Log duplicate unique_id collisions when classes differ."""
existing_class = self._entity_class_path(existing_entity)
duplicate_class = self._entity_class_path(duplicate_entity)
if existing_class == duplicate_class:
return

_LOGGER.warning(

@zigpy-review-bot zigpy-review-bot May 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_LOGGER.warning is the right level for a real conflict, but with _pending_entities never reset between async_initialize calls (gateway drives it multiple times: cache=True initial, cache=False after online, re-interview), this fires per call for the same device. The PR body's 4-7 repeats per IEEE confirm it.

Suggestion: gate on a per-device set[tuple[Platform, str]] so each (platform, unique_id) conflict is logged once per device lifetime. Keeps warning-level (which is right — conflicts are bugs), without N-spamming the same finding.

# in __init__:
self._logged_unique_id_conflicts: set[tuple[Platform, str]] = set()

# in _log_entity_unique_id_conflict, after the class-equality check:
if key in self._logged_unique_id_conflicts:
    return
self._logged_unique_id_conflicts.add(key)

"Entity unique_id conflict on device %s for %s: keeping %s "
"(endpoint=%s, cluster_handlers=%s), dropping %s "
"(endpoint=%s, cluster_handlers=%s)",
self.ieee,
key,
".".join(existing_class),
self._entity_endpoint_id(existing_entity),
self._entity_cluster_handler_names(existing_entity),
".".join(duplicate_class),
self._entity_endpoint_id(duplicate_entity),
self._entity_cluster_handler_names(duplicate_entity),
)

async def async_initialize(self, from_cache: bool = False) -> None:
"""Initialize cluster handlers."""
self.debug("started initialization")
Expand Down Expand Up @@ -1012,6 +1057,7 @@ async def async_initialize(self, from_cache: bool = False) -> None:

# Ignore entities that already exist
if key in new_entities:
self._log_entity_unique_id_conflict(key, new_entities[key], entity)

@zigpy-review-bot zigpy-review-bot May 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth confirming in the PR body: this is intentionally surface-the-bug logging, not a fix. The duplicate FirmwareUpdateServerEntity registration on Ota.cluster_id with the same _unique_id_suffix = "firmware_update" is the underlying issue; this PR just makes it visible. The actual fix presumably lives in giving the server entity a distinct suffix or scoping its @register_entity to coordinator devices only (since it's the OTA server side).

await entity.on_remove()
continue

Expand Down
Loading