@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+
+"""Browse for HTTP services with the synchronous API and print each event."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+
+from zeroconf import IPVersion, ServiceBrowser, ServiceListener, Zeroconf
+
+
+class PrintingListener(ServiceListener):
+    def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
+        info = zc.get_service_info(type_, name)
+        print(f"added: {name}")
+        if info is not None:
+            print(f"  addresses: {', '.join(info.parsed_scoped_addresses())}")
+            print(f"  server: {info.server} port: {info.port}")
+            print(f"  properties: {info.decoded_properties}")
+
+    def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None:
+        print(f"removed: {name}")
+
+    def update_service(self, zc: Zeroconf, type_: str, name: str) -> None:
+        print(f"updated: {name}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--debug", action="store_true", help="enable debug logging")
+    parser.add_argument("--v6-only", action="store_true", help="use IPv6 only")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
+    ip_version = IPVersion.V6Only if args.v6_only else IPVersion.All
+
+    zc = Zeroconf(ip_version=ip_version)
+    browser = ServiceBrowser(zc, "_http._tcp.local.", PrintingListener())
+    print("browsing for _http._tcp.local., press enter to exit")
+    try:
+        input()
+    finally:
+        zc.close()
@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+
+"""Register a demo HTTP service and keep it announced until interrupted."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import socket
+from time import sleep
+
+from zeroconf import IPVersion, ServiceInfo, Zeroconf
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--debug", action="store_true", help="enable debug logging")
+    parser.add_argument("--v6-only", action="store_true", help="use IPv6 only")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
+    ip_version = IPVersion.V6Only if args.v6_only else IPVersion.All
+
+    service = ServiceInfo(
+        "_http._tcp.local.",
+        "Demo Web Service._http._tcp.local.",
+        addresses=[socket.inet_aton("127.0.0.1")],
+        port=8080,
+        properties={"path": "/"},
+        server="demo-host.local.",
+    )
+
+    zc = Zeroconf(ip_version=ip_version)
+    print("registering Demo Web Service._http._tcp.local., press ctrl-c to exit")
+    zc.register_service(service)
+    try:
+        while True:
+            sleep(0.5)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        print("unregistering")
+        zc.unregister_service(service)
+        zc.close()
@@ -150,7 +150,8 @@ def async_send_with_transport(
 
 
 class Zeroconf(QuietLogger):
-
+    """High level mDNS service discovery: register services, browse for
+    them, and resolve their details on the local network."""
 
     def __init__(
         self,
@@ -160,7 +161,7 @@ def __init__(
         apple_p2p: bool = False,
         use_asyncio: bool | None = None,
     ) -> None:
-        multicast communications, listening and reaping threads.
+        """Open the multicast sockets and start serving on the chosen interfaces.
 
         :param interfaces: :class:`InterfaceChoice` or a list of IP addresses
             (IPv4 and IPv6) and interface indexes (IPv6 only).
@@ -317,17 +318,18 @@ def get_service_info(
         return None
 
     def add_service_listener(self, type_: str, listener: ServiceListener) -> None:
-        will then have its add_service and remove_service methods called when
-        services of that type become available and unavailable."""
+        """Browse for a type and deliver add, remove, and update callbacks to the listener."""
         self.remove_service_listener(listener)
         self.browsers[listener] = ServiceBrowser(self, type_, listener)
 
     def remove_service_listener(self, listener: ServiceListener) -> None:
+        """Stop the browser behind the given listener."""
         if listener in self.browsers:
             self.browsers[listener].cancel()
             del self.browsers[listener]
 
     def remove_all_service_listeners(self) -> None:
+        """Stop the browsers behind every registered listener."""
         for listener in list(self.browsers):
             self.remove_service_listener(listener)
 
@@ -647,6 +649,7 @@ async def async_check_service(
         cooperating_responders: bool = False,
         strict: bool = True,
     ) -> None:
+        """Probe the network for name uniqueness, renaming first when allowed."""
         instance_name = instance_name_from_service_info(info, strict=strict)
         if cooperating_responders:
             return
@@ -685,11 +688,7 @@ def add_listener(
         listener: RecordUpdateListener,
         question: DNSQuestion | list[DNSQuestion] | None,
     ) -> None:
-        its update_record method called when information is available to
-        answer the question(s).
-
-        This function is threadsafe
-        """
+        """Subscribe a record update listener, optionally scoped to questions; threadsafe."""
         assert self.loop is not None
         self.loop.call_soon_threadsafe(self.record_manager.async_add_listener, listener, question)
 
@@ -706,11 +705,7 @@ def async_add_listener(
         listener: RecordUpdateListener,
         question: DNSQuestion | list[DNSQuestion] | None,
     ) -> None:
-        its update_record method called when information is available to
-        answer the question(s).
-
-        This function is not threadsafe and must be called in the eventloop.
-        """
+        """Subscribe a record update listener, optionally scoped to questions; event loop only."""
         self.record_manager.async_add_listener(listener, question)
 
     def async_remove_listener(self, listener: RecordUpdateListener) -> None:
@@ -797,10 +792,7 @@ def _shutdown_threads(self) -> None:
         self.loop.close()
 
     def close(self) -> None:
-        servicing further queries.
-
-        This method is idempotent and irreversible.
-        """
+        """Shut down the sockets and engine for good; safe to call repeatedly."""
         assert self.loop is not None
         if self.loop.is_running():
             if self.loop == get_running_loop():
@@ -814,15 +806,7 @@ def close(self) -> None:
         self._shutdown_threads()
 
     async def _async_close(self) -> None:
-        servicing further queries.
-
-        This method is idempotent and irreversible.
-
-        This call only intended to be used by AsyncZeroconf
-
-        Callers are responsible for unregistering all services
-        before calling this function
-        """
+        """Shut down for AsyncZeroconf; callers unregister services first."""
         self._close()
         await self.engine._async_close()  # pylint: disable=protected-access
         self._shutdown_threads()
 
@@ -26,6 +26,7 @@
 import socket
 from typing import TYPE_CHECKING, Any, cast
 
+from ._exceptions import AbstractMethodException
 from ._utils.net import _is_v6_address
 from ._utils.time import current_time_millis
 from .const import _CLASS_MASK, _CLASS_UNIQUE, _CLASSES, _TYPE_ANY, _TYPES
@@ -156,7 +157,7 @@ def __repr__(self) -> str:
         )
 
 
-class DNSRecord(DNSEntry):
+class DNSRecord(DNSEntry):  # noqa: PLW1641
     """A DNS record - like a DNS entry, but has a TTL"""
 
     __slots__ = ("created", "ttl")
@@ -177,6 +178,9 @@ def _fast_init_record(self, name: str, type_: _int, class_: _int, ttl: _int, cre
         self.ttl = ttl
         self.created = created
 
+    def __eq__(self, other: Any) -> bool:
+        raise AbstractMethodException(f"{type(self).__name__} is missing __eq__")
+
     def __lt__(self, other: DNSRecord) -> bool:
         return self.ttl < other.ttl
 
@@ -187,6 +191,16 @@ def suppressed_by(self, msg: DNSIncoming) -> bool:
                 return True
         return False
 
+    def _suppressed_by_answer(self, answer: DNSRecord) -> bool:
+        """True when the answer matches this record with at least half its TTL left."""
+        return self == answer and self.ttl / 2 < answer.ttl
+
+    def get_expiration_time(self, percent: _int) -> float:
+        """Return the moment when the given percentage of the TTL has elapsed."""
+        # created is epoch milliseconds and ttl is seconds, so one percent
+        # of the ttl expressed in milliseconds is ttl * 10
+        return self.created + percent * self.ttl * 10
+
     # TODO: Switch to just int here
     def get_remaining_ttl(self, now: _float) -> int | float:
         """Returns the remaining TTL in seconds."""
@@ -211,6 +225,9 @@ def _set_created_ttl(self, created: _float, ttl: _int) -> None:
         self.created = created
         self.ttl = ttl
 
+    def write(self, out: DNSOutgoing) -> None:
+        raise AbstractMethodException(f"{type(self).__name__} is missing write")
+
     def to_string(self, other: bytes | str) -> str:
         arg = f"{self.ttl}/{int(self.get_remaining_ttl(current_time_millis()))},{cast(Any, other)}"
         return DNSEntry.entry_to_string(self, "record", arg)
 
@@ -53,32 +53,21 @@ def __init__(self, zeroconf: Zeroconf) -> None:
         self.listeners: set[RecordUpdateListener] = set()
 
     def async_updates(self, now: _float, records: list[_RecordUpdate]) -> None:
-        a record.
-
-        This method must be called before the cache is updated.
-
-        This method will be run in the event loop.
-        """
+        """Notify listeners of updated records, before the cache commits them."""
         for listener in self.listeners.copy():
             listener.async_update_records(self.zc, now, records)
 
     def async_updates_complete(self, notify: bool) -> None:
-        a record.
-
-        This method must be called after the cache is updated.
-
-        This method will be run in the event loop.
-        """
+        """Notify listeners that a batch landed, after the cache committed it."""
         for listener in self.listeners.copy():
             listener.async_update_records_complete()
         if notify:
             self.zc.async_notify_all()
 
     def async_updates_from_response(self, msg: DNSIncoming) -> None:
-        are held in the cache, and listeners are notified.
+        """Ingest a response: cache its answers and notify listeners.
 
-        This function must be run in the event loop as it is not
-        threadsafe.
+        Runs in the event loop; not threadsafe.
         """
         updates: list[RecordUpdate] = []
         address_adds: list[DNSRecord] = []
@@ -168,10 +157,8 @@ def async_add_listener(
         listener: RecordUpdateListener,
         question: DNSQuestion | list[DNSQuestion] | None,
     ) -> None:
-        its update_record method called when information is available to
-        answer the question(s).
-
-        This function is not thread-safe and must be called in the eventloop.
+        """Subscribe a listener, optionally scoped to questions, and seed it
+        with matching cached records; event loop only.
         """
         if not isinstance(listener, RecordUpdateListener):
             log.error(  # type: ignore[unreachable]
 
@@ -55,8 +55,8 @@
 
 
 class AsyncListener:
-
-    the read() method called when a socket is available for reading."""
+    """Asyncio datagram protocol that receives mDNS packets and feeds them
+    to the record manager and query handler as they arrive."""
 
     __slots__ = (
         "_deferred",
@@ -236,6 +236,7 @@ def handle_query_or_defer(
         transport: _WrappedTransport,
         v6_flow_scope: tuple[()] | tuple[int, int],
     ) -> None:
+        """Answer immediately, or hold a truncated query for reassembly."""
         if not msg.truncated:
             self._respond_query(msg, addr, port, transport, v6_flow_scope)
             return
 
@@ -311,6 +311,7 @@ def _read_string(self, length: _int) -> bytes:
         return info
 
     def _read_others(self) -> None:
+        """Read the answer, authority and additional sections."""
         self._did_read_others = True
         buf = self._buf
         answers = self._answers
 
@@ -65,10 +65,10 @@ cdef class DNSOutgoing:
     cdef cython.bint _write_question(self, DNSQuestion question)
 
     @cython.locals(
-        d=cython.bytes,
-        data_view=cython.list,
-        index=cython.uint,
-        length=cython.uint
+        data_length_before=cython.uint,
+        size_before=cython.uint,
+        length_index=cython.uint,
+        rdata_size_start=cython.uint,
     )
     cdef cython.bint _write_record(self, DNSRecord record, double now)
 
@@ -110,7 +110,7 @@ cdef class DNSOutgoing:
 
     cpdef void write_character_string(self, cython.bytes value)
 
-    @cython.locals(utfstr=bytes)
+    @cython.locals(encoded=bytes, byte_length=cython.uint)
     cdef void _write_utf(self, cython.str value)
 
     @cython.locals(
@@ -132,7 +132,6 @@ cdef class DNSOutgoing:
 
     cpdef void add_answer(self, DNSIncoming inp, DNSRecord record)
 
-    @cython.locals(now_double=double)
     cpdef void add_answer_at_time(self, DNSRecord record, double now)
 
     cpdef void add_authorative_answer(self, DNSPointer record)