feat(sqldata): standardize on synchronous direct SqlDataSocket and remove async TCP tunnel

- Replace local TCP loopback tunnel server and event loop reader/writer forwarding with direct synchronous SqlDataSocket implementation
- Remove async mode, GOOGLE_CLOUD_SQL_DATA_IMPL toggle, and FallbackSocket
- Update SqlDataClient to manage synchronous gRPC streams and return SqlDataSocket wrapping _RequestQueue and background response reader
- Update unit tests to verify direct socket connection and lifecycle

Author: hessjcg

google/cloud/sql/connector/connector.py

@@ -43,13 +43,13 @@
 from google.cloud.sql.connector.enums import RefreshStrategy
 from google.cloud.sql.connector.exceptions import ClosedConnectorError
 from google.cloud.sql.connector.exceptions import ConnectorLoopError
+from google.cloud.sql.connector.exceptions import IncompatibleDriverError
 from google.cloud.sql.connector.exceptions import ResourceExhaustedError
 from google.cloud.sql.connector.instance import RefreshAheadCache
 from google.cloud.sql.connector.lazy import LazyRefreshCache
 from google.cloud.sql.connector.monitored_cache import MonitoredCache
 from google.cloud.sql.connector.resolver import DefaultResolver
 from google.cloud.sql.connector.resolver import DnsResolver
-from google.cloud.sql.connector.sqldata_client import FallbackSocket
 from google.cloud.sql.connector.sqldata_client import SqlDataClient
 from google.cloud.sql.connector.utils import format_database_user
 from google.cloud.sql.connector.utils import generate_keys
@@ -283,7 +283,7 @@ def __init__(
         )
         self._sql_data_fallback_cache: set[str] = set()
         self._sql_data_conn_state: dict[str, SqlDataConnState] = {}
-        self._sqldata_clients: set[SqlDataClient] = set()
+        self._sqldata_clients: set[Any] = set()
 
 
 
@@ -468,6 +468,10 @@ async def connect_async(
         # attempt to establish connection
         try:
             if ip_type == IPTypes.SQL_DATA:
+                if driver in ASYNC_DRIVERS:
+                    raise IncompatibleDriverError(
+                        f"Driver '{driver}' is not supported with ip_type '{ip_type}'."
+                    )
                 state = self._sql_data_conn_state.setdefault(
                     str(conn_name), SqlDataConnState()
                 )
@@ -526,7 +530,7 @@ async def get_conn_info():
                     cache = self._get_or_create_cache(conn_name, enable_iam_auth)
                     return await cache.connect_info()
 
-                tunnel_port = await sqldata_client.connect_tunnel(
+                sock = await sqldata_client.connect(
                     instance_connection_name=str(conn_name),
                     region=conn_name.region,
                     project=conn_name.project,
@@ -539,36 +543,24 @@ async def get_conn_info():
                     connect_timeout=kwargs.get("timeout", self._timeout),
                 )
 
-                if driver in ASYNC_DRIVERS:
-                    return await connector(
-                        "127.0.0.1",
-                        None,
-                        port=tunnel_port,
-                        **kwargs,
-                    )
-                else:
-                    raw_sock = socket.create_connection(("127.0.0.1", tunnel_port))
-                    fd = raw_sock.detach()
-                    fallback_sock = FallbackSocket(fileno=fd)
-
-                    cache = None
-                    if conn_name.domain_name:
-                        cache = self._get_or_create_cache(conn_name, enable_iam_auth)
-                        cache.sockets.append(fallback_sock)
-
-                    connect_partial = partial(
-                        connector,
-                        "127.0.0.1",
-                        fallback_sock,
-                        **kwargs,
-                    )
-                    try:
-                        return await self._loop.run_in_executor(None, connect_partial)
-                    except Exception:
-                        fallback_sock.close()
-                        if conn_name.domain_name and cache:
-                            cache._purge_closed_sockets()
-                        raise
+                cache = None
+                if conn_name.domain_name:
+                    cache = self._get_or_create_cache(conn_name, enable_iam_auth)
+                    cache.sockets.append(sock)
+
+                connect_partial = partial(
+                    connector,
+                    "127.0.0.1",
+                    sock,
+                    **kwargs,
+                )
+                try:
+                    return await self._loop.run_in_executor(None, connect_partial)
+                except Exception:
+                    sock.close()
+                    if conn_name.domain_name and cache:
+                        cache._purge_closed_sockets()
+                    raise
             else:
                 try:
                     conn_info = await monitored_cache.connect_info()

google/cloud/sql/connector/proto/sql_data_service_pb2.py

@@ -34,6 +34,8 @@
 )
 # @@protoc_insertion_point(imports)
 
+from google.rpc import status_pb2 as _status_pb2  # noqa: F401
+
 _sym_db = _symbol_database.Default()