[karenc-bq]

Description

The async topology monitor held the provider that built it and called back into it, so every provider ran its own background task, topology cache and monitoring connection for the same cluster. This puts the async path on the shared storage service and monitor service, as the synchronous provider already is.

1. The monitor renewed its own expiration and could never be reaped as idle. A storage read publishes a data access, which the monitor service turns into an expiration extension for the monitor producing that type, and the monitor reads the topology it produces every tick. StorageService.get gains a keyword-only register_data_access flag, default True, and the monitor's own read passes False. The expiration is still unreachable on its own because can_dispose is True only after the monitor stops; that is tracked separately.
2. Async topology lived in a per-provider cache, so a second connection to a cluster started cold. It now lives in the storage service under a (cluster_id, loop_token) key. The loop token is needed because entries hold mutable HostInfo objects written in place, and a monitor owns a task bound to one loop, so a bare cluster id would hand a fresh loop a dead loop's entry. Readers get copies.
3. Each provider ran its own monitor. The monitor now takes a topology-utils object, holding one dialect's query and its row parsing, rather than a provider, so one monitor serves every provider on a cluster and loop. The provider gets it from MonitorService.run_if_absent under the same key the topology is stored at. MonitorService has no liveness check and extends expiration on every cache hit, so a monitor on a closed loop would be handed out until the stuck sweep removed it while every blocking refresh waited for an update that could not arrive. The provider rechecks is_running and replaces a dead monitor once, removing it by identity through the new stop_and_remove_value, because a cleanup pass may already have put a replacement at the key.
4. The cluster id was derived from the endpoint. It is now the cluster_id property, default "1", matching the synchronous provider and the property's own documentation. Connections sharing an id share a topology entry and a monitor, so an application using more than one cluster has to set it per cluster.

Behaviour changes

1. refresh() no longer applies its own freshness window; it returns the stored entry and the monitor keeps it fresh. topology_refresh_ms still sets the monitor's tick interval.
2. An invalid cluster_instance_host_pattern is rejected when the provider is built, during connect, rather than on the first topology query. That is where the synchronous provider validates it.
3. A blue/green reset now clears the async topology. Nothing consumed MonitorResetEvent on the async side before, so a switchover left the old host list readable until it expired.

Deliberate divergences from sync

1. stop() is non-blocking and schedules teardown on the monitor's own loop, where sync joins its thread for 30 seconds. MonitorStopEvent is delivered on the publisher's thread and async blue/green publishes from a coroutine, so blocking would wedge that loop. An awaitable aclose() stays registered as a shutdown hook, since release_resources clears the monitor cache and would drop the last reference first.
2. can_dispose reports whether a refresh is in flight. Sync reports self._stop.is_set(), false for every live monitor, which makes its expiration dead code. Filed separately.
3. process_event ends by removing the stored topology, which sync does not do. Filed separately.

Other Fixes

1. A cluster reached on a non-default port produced topology hosts on the dialect default port.
2. The Global Aurora parser kept every row claiming the writer role, where a cross-region switchover can report two. ExpirationTrackingCache.detach_value reported success whether or not it removed anything.
3. tests/unit/conftest.py gains a monitor-service reset, since monitors are keyed partly by loop and a recycled loop identity let one test's monitor reach the next.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.