@@ -58,3 +58,11 @@ repos:
     hooks:
       - id: cython-lint
       - id: double-quote-cython-strings
+  - repo: local
+    hooks:
+      - id: check-declaration-order
+        name: check class declaration order
+        entry: python scripts/check_declaration_order.py
+        language: system
+        pass_filenames: false
+        files: ^src/zeroconf/.*\.py$
@@ -183,6 +183,9 @@ select = [
 "examples/**/*" = [
     "T201", # intended
 ]
+"scripts/**/*" = [
+    "T201", # intended
+]
 "setup.py" = ["D100"]
 "conftest.py" = ["D100"]
 "docs/conf.py" = ["D100"]
 
@@ -0,0 +1,50 @@
+"""Enforce the repository's class declaration ordering convention."""
+
+from __future__ import annotations
+
+import ast
+import sys
+from pathlib import Path
+
+SRC = Path(__file__).resolve().parent.parent / "src" / "zeroconf"
+
+
+def sort_key(name: str) -> tuple[int, int, str]:
+    is_dunder = name.startswith("__") and name.endswith("__")
+    is_private = not is_dunder and name.startswith("_")
+    group = 0 if is_dunder else (2 if is_private else 1)
+    return (group, 0 if name == "__init__" else 1, name)
+
+
+def check_class(path: Path, cls: ast.ClassDef) -> list[str]:
+    names = [node.name for node in cls.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))]
+    expected = sorted(names, key=sort_key)
+    if names == expected:
+        return []
+    # property setters share the getter's name; a stable sort keeps their
+    # relative order, so equal-name adjacency is always satisfiable
+    for actual, want in zip(names, expected, strict=True):
+        if actual != want:
+            return [
+                f"{path}:{cls.lineno}: class {cls.name}: method order breaks the "
+                f"convention at {actual!r} (expected {want!r}); required order is "
+                f"dunders with __init__ first, then public, then private, each "
+                f"alphabetical"
+            ]
+    return []
+
+
+def main() -> int:
+    problems: list[str] = []
+    for path in sorted(SRC.rglob("*.py")):
+        tree = ast.parse(path.read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ClassDef):
+                problems.extend(check_class(path.relative_to(SRC.parent.parent), node))
+    for problem in problems:
+        print(problem)
+    return 1 if problems else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -117,68 +117,6 @@ def __init__(
         self._cleanup_timer: asyncio.TimerHandle | None = None
         self._setup_task: asyncio.Task[None] | None = None
 
-    def setup(
-        self,
-        loop: asyncio.AbstractEventLoop,
-        loop_thread_ready: threading.Event | None,
-    ) -> None:
-        """Set up the instance."""
-        self.loop = loop
-        self.running_future = loop.create_future()
-        self._setup_task = self.loop.create_task(self._async_setup(loop_thread_ready))
-
-    async def _async_setup(self, loop_thread_ready: threading.Event | None) -> None:
-        """Set up the instance."""
-        self._async_schedule_next_cache_cleanup()
-        await self._async_create_endpoints()
-        assert self.running_future is not None
-        if not self.running_future.done():
-            self.running_future.set_result(True)
-        if loop_thread_ready:
-            loop_thread_ready.set()
-
-    async def _async_create_endpoints(self) -> None:
-        """Create endpoints to send and receive."""
-        reader_sockets = []
-        sender_sockets = []
-        if self._listen_socket:
-            reader_sockets.append(self._listen_socket)
-        for s in self._respond_sockets:
-            if s not in reader_sockets:
-                reader_sockets.append(s)
-            sender_sockets.append(s)
-
-        for s in reader_sockets:
-            reader = await self._async_wrap_socket(s, s in sender_sockets)
-            # _async_wrap_socket registers the transport with no await between
-            # creating and registering it, and the pending-handle cleanup below
-            # adds no await either, so a concurrent shutdown always sees ``s``
-            # in exactly one place.
-            if s is self._listen_socket:
-                # Keep a handle to the shared listen socket so interface
-                # rescans can add/drop multicast memberships on it.
-                self._listen_transport = reader
-                self._listen_socket = None
-            if s in self._respond_sockets:
-                self._respond_sockets.remove(s)
-
-    async def _async_wrap_socket(self, sock: socket.socket, is_sender: bool) -> _WrappedTransport:
-        """Adopt a socket into a transport, register it, and return the reader wrapper."""
-        assert self.loop is not None
-        transport, protocol = await self.loop.create_datagram_endpoint(  # type: ignore[type-var]
-            lambda: AsyncListener(self.zc),  # type: ignore[arg-type, return-value]
-            sock=sock,
-        )
-        datagram_transport = cast(asyncio.DatagramTransport, transport)
-        reader = make_wrapped_transport(datagram_transport)
-        # No ``await`` between wrapping and registering so a concurrent
-        # shutdown always sees the transport in exactly one place.
-        self.protocols.append(cast(AsyncListener, protocol))
-        self.readers.append(reader)
-        if is_sender:
-            self.senders.append(make_wrapped_transport(datagram_transport))
-        return reader
-
     async def async_update_interfaces(
         self,
         interfaces: InterfacesType,
@@ -271,6 +209,32 @@ async def async_update_interfaces(
                 added = True
         return added
 
+    def close(self) -> None:
+        """Close from sync context.
+
+        While it is not expected during normal operation,
+        this function may raise EventLoopBlocked if the underlying
+        call to `_async_close` cannot be completed.
+        """
+        assert self.loop is not None
+        # Guard against Zeroconf.close() being called from the eventloop
+        if get_running_loop() == self.loop:
+            self._async_shutdown()
+            return
+        if not self.loop.is_running():
+            return
+        run_coro_with_timeout(self._async_close(), self.loop, _CLOSE_TIMEOUT)
+
+    def setup(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        loop_thread_ready: threading.Event | None,
+    ) -> None:
+        """Set up the instance."""
+        self.loop = loop
+        self.running_future = loop.create_future()
+        self._setup_task = self.loop.create_task(self._async_setup(loop_thread_ready))
+
     async def _async_add_interface(
         self,
         interface: str | tuple[tuple[str, int, int], int],
@@ -309,19 +273,32 @@ async def _async_add_interface(
             return False
         return True
 
-    def _async_remove_transport(self, transport: asyncio.DatagramTransport) -> None:
-        """Drop a transport's protocol/reader/sender wrappers, cancelling its timers."""
-        kept_protocols = []
-        for protocol in self.protocols:
-            if protocol.transport is not None and protocol.transport.transport is transport:
-                # Cancel any pending TC-reassembly timers so one can't fire a
-                # response against the transport we're about to close.
-                protocol.cancel_pending_timers()
-            else:
-                kept_protocols.append(protocol)
-        self.protocols = kept_protocols
-        self.readers = _without_transport(self.readers, transport)
-        self.senders = _without_transport(self.senders, transport)
+    def _async_cache_cleanup(self) -> None:
+        """Periodic cache cleanup."""
+        now = current_time_millis()
+        self.zc.question_history.async_expire(now)
+        self.zc.record_manager.async_updates(
+            now,
+            [RecordUpdate(record, record) for record in self.zc.cache.async_expire(now)],
+        )
+        self.zc.record_manager.async_updates_complete(False)
+        self._async_schedule_next_cache_cleanup()
+
+    async def _async_close(self) -> None:
+        """Cancel and wait for the cleanup task to finish."""
+        assert self._setup_task is not None
+        # Swallow CancelledError only if the setup task itself was
+        # cancelled (close-before-start); outer-task cancellation must
+        # propagate.
+        try:
+            await self._setup_task
+        except asyncio.CancelledError:
+            if not self._setup_task.cancelled():
+                raise
+        self._async_shutdown()
+        await asyncio.sleep(0)  # flush out any call soons
+        if self._cleanup_timer is not None:
+            self._cleanup_timer.cancel()
 
     def _async_close_sender(self, wrapped: _WrappedTransport, listen_socket: socket.socket | None) -> None:
         """Drop a per-interface sender's wrappers/protocol and close its transport."""
@@ -334,6 +311,31 @@ def _async_close_sender(self, wrapped: _WrappedTransport, listen_socket: socket.
             # Release the socket even if a non-benign leave (e.g. EPERM) raises.
             transport.close()
 
+    async def _async_create_endpoints(self) -> None:
+        """Create endpoints to send and receive."""
+        reader_sockets = []
+        sender_sockets = []
+        if self._listen_socket:
+            reader_sockets.append(self._listen_socket)
+        for s in self._respond_sockets:
+            if s not in reader_sockets:
+                reader_sockets.append(s)
+            sender_sockets.append(s)
+
+        for s in reader_sockets:
+            reader = await self._async_wrap_socket(s, s in sender_sockets)
+            # _async_wrap_socket registers the transport with no await between
+            # creating and registering it, and the pending-handle cleanup below
+            # adds no await either, so a concurrent shutdown always sees ``s``
+            # in exactly one place.
+            if s is self._listen_socket:
+                # Keep a handle to the shared listen socket so interface
+                # rescans can add/drop multicast memberships on it.
+                self._listen_transport = reader
+                self._listen_socket = None
+            if s in self._respond_sockets:
+                self._respond_sockets.remove(s)
+
     async def _async_rebuild_listen_socket(
         self,
         apple_p2p: bool,
@@ -387,38 +389,35 @@ async def _async_rebuild_listen_socket(
         self._async_remove_transport(old_transport)
         old_transport.close()
 
-    def _async_cache_cleanup(self) -> None:
-        """Periodic cache cleanup."""
-        now = current_time_millis()
-        self.zc.question_history.async_expire(now)
-        self.zc.record_manager.async_updates(
-            now,
-            [RecordUpdate(record, record) for record in self.zc.cache.async_expire(now)],
-        )
-        self.zc.record_manager.async_updates_complete(False)
-        self._async_schedule_next_cache_cleanup()
+    def _async_remove_transport(self, transport: asyncio.DatagramTransport) -> None:
+        """Drop a transport's protocol/reader/sender wrappers, cancelling its timers."""
+        kept_protocols = []
+        for protocol in self.protocols:
+            if protocol.transport is not None and protocol.transport.transport is transport:
+                # Cancel any pending TC-reassembly timers so one can't fire a
+                # response against the transport we're about to close.
+                protocol.cancel_pending_timers()
+            else:
+                kept_protocols.append(protocol)
+        self.protocols = kept_protocols
+        self.readers = _without_transport(self.readers, transport)
+        self.senders = _without_transport(self.senders, transport)
 
     def _async_schedule_next_cache_cleanup(self) -> None:
         """Schedule the next cache cleanup."""
         loop = self.loop
         assert loop is not None
         self._cleanup_timer = loop.call_at(loop.time() + _CACHE_CLEANUP_INTERVAL, self._async_cache_cleanup)
 
-    async def _async_close(self) -> None:
-        """Cancel and wait for the cleanup task to finish."""
-        assert self._setup_task is not None
-        # Swallow CancelledError only if the setup task itself was
-        # cancelled (close-before-start); outer-task cancellation must
-        # propagate.
-        try:
-            await self._setup_task
-        except asyncio.CancelledError:
-            if not self._setup_task.cancelled():
-                raise
-        self._async_shutdown()
-        await asyncio.sleep(0)  # flush out any call soons
-        if self._cleanup_timer is not None:
-            self._cleanup_timer.cancel()
+    async def _async_setup(self, loop_thread_ready: threading.Event | None) -> None:
+        """Set up the instance."""
+        self._async_schedule_next_cache_cleanup()
+        await self._async_create_endpoints()
+        assert self.running_future is not None
+        if not self.running_future.done():
+            self.running_future.set_result(True)
+        if loop_thread_ready:
+            loop_thread_ready.set()
 
     def _async_shutdown(self) -> None:
         """Shutdown transports and sockets; safe to call repeatedly."""
@@ -439,18 +438,19 @@ def _async_shutdown(self) -> None:
             s.close()
         self._respond_sockets = []
 
-    def close(self) -> None:
-        """Close from sync context.
-
-        While it is not expected during normal operation,
-        this function may raise EventLoopBlocked if the underlying
-        call to `_async_close` cannot be completed.
-        """
+    async def _async_wrap_socket(self, sock: socket.socket, is_sender: bool) -> _WrappedTransport:
+        """Adopt a socket into a transport, register it, and return the reader wrapper."""
         assert self.loop is not None
-        # Guard against Zeroconf.close() being called from the eventloop
-        if get_running_loop() == self.loop:
-            self._async_shutdown()
-            return
-        if not self.loop.is_running():
-            return
-        run_coro_with_timeout(self._async_close(), self.loop, _CLOSE_TIMEOUT)
+        transport, protocol = await self.loop.create_datagram_endpoint(  # type: ignore[type-var]
+            lambda: AsyncListener(self.zc),  # type: ignore[arg-type, return-value]
+            sock=sock,
+        )
+        datagram_transport = cast(asyncio.DatagramTransport, transport)
+        reader = make_wrapped_transport(datagram_transport)
+        # No ``await`` between wrapping and registering so a concurrent
+        # shutdown always sees the transport in exactly one place.
+        self.protocols.append(cast(AsyncListener, protocol))
+        self.readers.append(reader)
+        if is_sender:
+            self.senders.append(make_wrapped_transport(datagram_transport))
+        return reader
@@ -73,12 +73,6 @@ def async_add(self, now: _float, answers: _AnswerWithAdditionalsType) -> None:
             loop.call_at(loop.time() + millis_to_seconds(random_delay), self.async_ready)
         self.queue.append(AnswerGroup(send_after, send_before, answers))
 
-    def _remove_answers_from_queue(self, answers: _AnswerWithAdditionalsType) -> None:
-        """Remove a set of answers from the outgoing queue."""
-        for pending in self.queue:
-            for record in answers:
-                pending.answers.pop(record, None)
-
     def async_ready(self) -> None:
         """Process anything in the queue that is ready."""
         zc = self.zc
@@ -113,3 +107,9 @@ def async_ready(self) -> None:
             # If we have the same answer scheduled to go out, remove them
             self._remove_answers_from_queue(answers)
             zc.async_send(construct_outgoing_multicast_answers(answers))
+
+    def _remove_answers_from_queue(self, answers: _AnswerWithAdditionalsType) -> None:
+        """Remove a set of answers from the outgoing queue."""
+        for pending in self.queue:
+            for record in answers:
+                pending.answers.pop(record, None)