fix: build the host list provider from a verified database dialect

aws_advanced_python_wrapper/aio/aurora_connection_tracker.py

@@ -602,7 +602,15 @@ def _parse_alias(alias: str) -> tuple:
         ``[::1]:5432`` split correctly on the LAST colon. Falls back to
         port 5432 when the alias has no colon (or the port part isn't
         numeric).
+
+        The trailing slash is stripped first: host-list-change keys are
+        ``HostInfo.url``, which is ``host:port/``. Without this, ``int()``
+        fails on ``"3306/"`` and the port silently became 5432, so a
+        rebuilt key never matched the tracked one and a demoted MySQL
+        writer's connections were never invalidated.
         """
+        if alias.endswith("/"):
+            alias = alias[:-1]
         host_part, sep, port_part = alias.rpartition(":")
         if not sep:
             # No colon -> treat entire alias as host, port defaults.

aws_advanced_python_wrapper/aio/dialect_utils.py

@@ -171,6 +171,33 @@ async def _execute_scalar_query(
                 except Exception:  # noqa: BLE001 - cursor close best-effort
                     pass
 
+    @staticmethod
+    async def fetch_one(conn: Any, query: str) -> Optional[tuple]:
+        """Run the provided query and fetch the first row.
+        """
+        return await AsyncDialectUtils._execute_scalar_query(conn, query)
+
+    @staticmethod
+    async def fetch_all(conn: Any, query: str) -> list:
+        """Run the provided query and fetch all rows. Returns an empty list on no results.
+        """
+        cursor_ret = conn.cursor()
+        cursor = await cursor_ret if asyncio.iscoroutine(cursor_ret) else cursor_ret
+        try:
+            await cursor.execute(query)
+            return list(await cursor.fetchall())
+        finally:
+            # Call close on the curosor if it has a close method.
+            # Some async DBAPI implementations may not have a close method on their cursor.
+            close = getattr(cursor, "close", None)
+            if close is not None:
+                try:
+                    result = close()
+                    if asyncio.iscoroutine(result):
+                        await result
+                except Exception:  # noqa: BLE001 - cursor close best-effort
+                    pass
+
     @staticmethod
     async def _execute_is_reader_query(
             conn: Any,

aws_advanced_python_wrapper/aio/gdb_failover_plugin.py

@@ -260,8 +260,8 @@ async def _failover_with_mode(
                 # deadline (validated on Aurora: 2592 reconnects to the stale
                 # host over 5 min, then timeout).
                 # STRICT_WRITER means "reconnect to the new writer" -- exactly
-                # failover_v2's job. (Requires the real topology provider, i.e.
-                # gdb_failover present in wrapper._TOPOLOGY_REQUIRING_PLUGINS.)
+                # failover_v2's job. (Requires a real topology provider, which
+                # the Global Aurora dialects supply.)
                 await self._failover_writer(
                     self._plugin_service.all_hosts, driver_dialect, deadline, None)
             case GdbFailoverMode.STRICT_HOME_READER:

aws_advanced_python_wrapper/aio/host_list_provider.py

@@ -141,6 +141,13 @@ def initial_connection_host_info(self, value: HostInfo) -> None:
     def is_static_host_list_provider(self) -> bool:
         ...
 
+    @property
+    def is_dialect_confirmed(self) -> bool:
+        ...
+
+    def get_target_driver_func(self) -> Optional[Callable]:
+        ...
+
 
 class AsyncStaticHostListProvider:
     """Static host list -- one host built from connection props.
@@ -221,7 +228,9 @@ def __init__(
             cluster_id: Optional[str] = None,
             default_port: int = 5432,
             monitor_connection_factory: Optional[
-                Callable[[], Awaitable[Any]]] = None) -> None:
+                Callable[[], Awaitable[Any]]] = None,
+            provider_service: Optional[AsyncHostListProviderService] = None) -> None:
+        self._provider_service = provider_service
         self._props = props
         self._driver_dialect = driver_dialect
         self._topology_query = topology_query
@@ -247,6 +256,22 @@ def __init__(
         # the wrapper once the plugin service exists -- see
         # aio/wrapper.py and cluster_topology_monitor.build_probe_host.
         self._probe_host: Optional[Callable[..., Any]] = None
+        self._instance_host_pattern, self._instance_pattern_port = \
+            self._resolve_instance_pattern()
+        # Resolve initial host
+        port_raw = props.get("port")
+        self._initial_hosts: Topology = (
+            HostInfo(
+                host=props.get("host", ""),
+                port=int(port_raw) if port_raw is not None else default_port,
+                role=HostRole.WRITER),)
+
+    def _is_dialect_confirmed(self) -> bool:
+        """Whether the database dialect is finalized and whether topology may be queried yet.
+        """
+        if self._provider_service is None:
+            return True
+        return self._provider_service.is_dialect_confirmed
 
     def set_probe_host(self, probe_host: Callable[..., Any]) -> None:
         """Arm panic-mode writer discovery on the (future) topology monitor.
@@ -258,25 +283,6 @@ def set_probe_host(self, probe_host: Callable[..., Any]) -> None:
         """
         self._probe_host = probe_host
 
-    def reconfigure_topology_query(
-            self, topology_query: str, default_port: Optional[int] = None) -> None:
-        """Re-point the provider at a (post-connect) resolved dialect's query.
-
-        The provider is built BEFORE the connection exists, so an
-        instance-endpoint MySQL connection starts with the PostgreSQL-default
-        topology query + port (the ctor default). Once the database dialect is
-        resolved/upgraded to the Aurora MySQL dialect we swap in its topology
-        query and the MySQL port so discovery actually works. Clears the cache
-        so the next refresh re-probes with the new query (any cached PG-query
-        result is stale/empty).
-        """
-        if topology_query:
-            self._topology_query = topology_query
-        if default_port:
-            self._default_port = default_port
-        self._topology_cache = None
-        self._last_refresh_ns = 0
-
     @staticmethod
     def _derive_cluster_id(props: Properties) -> str:
         # Only treat `cluster_id` as explicit when present in the props dict;
@@ -325,6 +331,11 @@ async def force_refresh(self, connection: Optional[Any] = None) -> Topology:
                 return self._topology_cache
             return ()
 
+        if not self._is_dialect_confirmed():
+            # No topology query and no monitor until the dialect is
+            # confirmed. Report the host from the connection properties in the meantime.
+            return self._topology_cache or self._initial_hosts
+
         async with self._refresh_lock:
             monitor = self._get_or_create_monitor()
             if monitor is None:
@@ -366,6 +377,10 @@ async def force_monitoring_refresh(
         querying through a caller connection. Falls back to
         :meth:`force_refresh` when no monitor can be built (static dialects).
         """
+        if not self._is_dialect_confirmed():
+            # No topology query and no monitor until the dialect is
+            # confirmed. Report the host from the connection properties in the meantime.
+            return self._topology_cache or self._initial_hosts
         monitor = self._get_or_create_monitor()
         if monitor is None:
             return await self.force_refresh(self._last_conn)
@@ -507,7 +522,7 @@ def _rows_to_topology(self, rows: List[tuple]) -> Topology:
         # The instance host pattern may carry an explicit ``:port`` (proxied
         # test endpoints do); use it for every topology host. See
         # :meth:`_resolve_instance_pattern`.
-        _, pattern_port = self._resolve_instance_pattern()
+        pattern_port = self._instance_pattern_port
         port = pattern_port if pattern_port is not None else self._default_port
         # Dedup by resolved host (a row may repeat), matching sync
         # AuroraTopologyUtils._process_query_results (:603-606).
@@ -573,6 +588,11 @@ def _rows_to_topology(self, rows: List[tuple]) -> Topology:
     def _resolve_instance_pattern(self) -> Tuple[Optional[str], Optional[int]]:
         """Resolve the instance host pattern and its explicit port (if any).
 
+        Called once, from ``__init__``; read afterwards through
+        ``_instance_host_pattern`` / ``_instance_pattern_port``. Keep it that
+        way -- it validates (and so can raise) and builds ``RdsUtils``, which
+        is why it does not belong on the per-row path.
+
         Returns ``(host_pattern_with_'?', port)``; ``port`` is ``None`` when the
         pattern carries no explicit ``:port`` suffix.
 
@@ -645,7 +665,7 @@ def _host_from_server_id(self, server_id: str) -> str:
         ID yields a non-resolvable name, so every reader/failover connect to a
         topology host fails with "failed to resolve host '<server_id>'".
         """
-        host_pattern, _ = self._resolve_instance_pattern()
+        host_pattern = self._instance_host_pattern
         if host_pattern:
             return host_pattern.replace("?", server_id)
         return server_id
@@ -708,6 +728,7 @@ def __init__(
             monitor_connection_factory: Optional[
                 Callable[[], Awaitable[Any]]] = None,
             writer_host_column_index: int = 0,
+            provider_service: Optional[AsyncHostListProviderService] = None,
     ) -> None:
         super().__init__(
             props=props,
@@ -716,6 +737,7 @@ def __init__(
             cluster_id=cluster_id,
             default_port=default_port,
             monitor_connection_factory=monitor_connection_factory,
+            provider_service=provider_service,
         )
         self._writer_host_query = writer_host_query
         self._host_id_query = host_id_query
@@ -859,6 +881,7 @@ def __init__(
             default_port: int = 5432,
             monitor_connection_factory: Optional[
                 Callable[[], Awaitable[Any]]] = None,
+            provider_service: Optional[AsyncHostListProviderService] = None,
     ) -> None:
         super().__init__(
             props=props,
@@ -867,6 +890,7 @@ def __init__(
             cluster_id=cluster_id,
             default_port=default_port,
             monitor_connection_factory=monitor_connection_factory,
+            provider_service=provider_service,
         )
         if instance_templates_by_region is None:
             raw = WrapperProperties.GLOBAL_CLUSTER_INSTANCE_HOST_PATTERNS.get(
@@ -954,3 +978,97 @@ def _rows_to_topology(self, rows: List[tuple]) -> Topology:
         # Writer first, readers after -- matches the other async providers.
         hosts.sort(key=lambda h: 0 if h.role == HostRole.WRITER else 1)
         return tuple(hosts)
+
+
+def build_monitor_connection_factory(
+        provider_service: AsyncHostListProviderService,
+        props: Properties) -> Callable[[], Awaitable[Any]]:
+    """Build the factory the topology monitor uses for its own connection.
+    """
+    from aws_advanced_python_wrapper.utils.properties import (Properties,
+                                                              PropertiesUtils)
+
+    async def _connect() -> Any:
+        target_func = provider_service.get_target_driver_func()
+        if target_func is None:
+            raise AwsWrapperError(
+                Messages.get_formatted(
+                    "HostListProvider.MonitorConnectionNoTargetDriverFunc",
+                    type(provider_service).__name__))
+        host_info = (provider_service.initial_connection_host_info
+                     or provider_service.current_host_info)
+        if host_info is None:
+            raise AwsWrapperError(
+                Messages.get_formatted(
+                    "HostListProvider.MonitorConnectionNoSeedHost",
+                    type(provider_service).__name__))
+        monitoring_props = PropertiesUtils.create_topology_monitoring_properties(
+            Properties(dict(props)))
+        return await provider_service.driver_dialect.connect(
+            host_info, monitoring_props, target_func)
+
+    return _connect
+
+
+def create_static_provider(
+        provider_service: AsyncHostListProviderService,
+        props: Properties) -> AsyncHostListProvider:
+    """Supplier body for dialects with no topology: a single seed host."""
+    return AsyncStaticHostListProvider(props)
+
+
+def create_aurora_provider(
+        provider_service: AsyncHostListProviderService,
+        props: Properties,
+        *,
+        topology_query: str,
+        default_port: int) -> AsyncHostListProvider:
+    """Supplier body for the Aurora topology dialects."""
+    return AsyncAuroraHostListProvider(
+        props,
+        provider_service.driver_dialect,
+        topology_query=topology_query,
+        default_port=default_port,
+        monitor_connection_factory=build_monitor_connection_factory(
+            provider_service, props),
+        provider_service=provider_service)
+
+
+def create_global_aurora_provider(
+        provider_service: AsyncHostListProviderService,
+        props: Properties,
+        *,
+        topology_query: str,
+        default_port: int) -> AsyncHostListProvider:
+    """Supplier body for the Aurora Global Database dialects."""
+    return AsyncGlobalAuroraHostListProvider(
+        props,
+        provider_service.driver_dialect,
+        topology_query=topology_query,
+        default_port=default_port,
+        monitor_connection_factory=build_monitor_connection_factory(
+            provider_service, props),
+        provider_service=provider_service)
+
+
+def create_multi_az_provider(
+        provider_service: AsyncHostListProviderService,
+        props: Properties,
+        *,
+        writer_host_query: str,
+        topology_query: str,
+        host_id_query: str,
+        default_port: int,
+        writer_host_column_index: int = 0) -> AsyncHostListProvider:
+    """Supplier body for the RDS Multi-AZ DB cluster dialects."""
+    return AsyncMultiAzHostListProvider(
+        props,
+        provider_service.driver_dialect,
+        writer_host_query=writer_host_query,
+        topology_query=topology_query,
+        host_id_query=host_id_query,
+        default_port=default_port,
+        writer_host_column_index=writer_host_column_index,
+        monitor_connection_factory=build_monitor_connection_factory(
+            provider_service, props),
+        provider_service=provider_service)