From 7f7e1909894aa97f9315bf06cb970169077cfb4a Mon Sep 17 00:00:00 2001
From: Mattie Fu <mattiefu@google.com>
Date: Tue, 20 Oct 2026 16:15:54 +0000
Subject: [PATCH 1/3] feat(bigtable): support the accelerator on Windows x86_64
 and x86

Change-Id: I7a17dd9919dff6e9d38b0cdfb3c600ffc363900c
---
 .../bigtable/data/_accelerator/_daemon.py     |  50 +++++++-
 .../data/_async/_accelerator_client.py        |  14 ++-
 .../cloud/bigtable/data/_async/client.py      |   8 --
 .../data/_sync_autogen/_accelerator_client.py |  13 ++-
 .../bigtable/data/_sync_autogen/client.py     |   7 --
 .../data/_async/test__accelerator_health.py   |   4 +-
 .../_sync_autogen/test__accelerator_health.py |   2 +-
 .../unit/data/test_accelerator_daemon.py      | 109 +++++++++++++++---
 8 files changed, 168 insertions(+), 39 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py
index 5e341d9c76fa..e3fbd9cb24ed 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py
@@ -30,10 +30,18 @@
 import shutil
 import socket
 import subprocess
+import sys
 import tempfile
 import time
 from typing import Any, Mapping, Sequence
 
+# Windows differs from every other supported platform in two ways that matter
+# here, so it gets a named constant rather than scattered ``sys.platform``
+# checks: the daemon binary carries a ``.exe`` suffix, and CPython's socket
+# module exposes no ``AF_UNIX`` even though the OS and gRPC's C-core both speak
+# it (see ``_socket_is_live``).
+_IS_WINDOWS = sys.platform == "win32"
+
 # Environment variable that overrides the daemon startup readiness timeout, in
 # seconds (float). Lets a deployment tune the cold-start budget without a code
 # change or rebuild — e.g. raise it on a very constrained/gVisor host, or (when
@@ -42,7 +50,10 @@
 _STARTUP_TIMEOUT_ENV_VAR = "BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT"
 
 # Wheels ship the binary at this path relative to the `_accelerator/` package.
+# The Windows wheels (win_amd64, win32) stage it under the ``.exe`` name Go
+# gives it, so the suffix is appended there.
 _DEFAULT_BIN_RELATIVE_PATH = "bin/accelerator"
+_WINDOWS_BIN_SUFFIX = ".exe"
 
 # The daemon writes the principal it resolved to this file in its tempdir
 # (alongside the socket) before binding, so the client can verify it matches
@@ -90,7 +101,8 @@
 
 def _resolve_binary_path() -> str:
     """Return the path of the bundled daemon binary, raising if it is absent."""
-    bundled = os.path.join(os.path.dirname(__file__), _DEFAULT_BIN_RELATIVE_PATH)
+    relative = _DEFAULT_BIN_RELATIVE_PATH + (_WINDOWS_BIN_SUFFIX if _IS_WINDOWS else "")
+    bundled = os.path.join(os.path.dirname(__file__), relative)
     if not os.path.isfile(bundled):
         raise FileNotFoundError(
             "Accelerator binary not found. Install a wheel that bundles the binary."
@@ -134,6 +146,23 @@ def _socket_is_live(uds_path: str) -> bool:
     """
     if not os.path.exists(uds_path):
         return False
+    if not hasattr(socket, "AF_UNIX"):
+        # Windows. The OS (10 1803+ / Server 2019+) and gRPC's C-core both
+        # support AF_UNIX, which is why the accelerator runs here at all, but
+        # CPython's socket module still does not expose the family, so we
+        # cannot open a probe connection from Python. The gate is
+        # ``HAVE_SYS_UN_H``: ``PC/pyconfig.h`` leaves it commented out and
+        # ``Modules/socketmodule.h`` responds with ``#undef AF_UNIX``, so the
+        # attribute is absent on every current CPython for Windows. Testing
+        # the attribute rather than the platform keeps this correct for any
+        # other build that lacks the family, and is what lets the tests
+        # exercise this branch from a POSIX run.
+        #
+        # Fall back to the bind artifact: the daemon's
+        # ``net.Listen("unix", ...)`` creates this file and puts the socket in
+        # the listen backlog inside one call, so the path existing means the
+        # listener is up.
+        return True
     with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
         probe.settimeout(_SOCKET_PROBE_TIMEOUT_SECONDS)
         try:
@@ -245,11 +274,21 @@ def start(self) -> None:
         """
         if self._proc is not None:
             raise RuntimeError("AcceleratorDaemon.start() called twice")
-        if not hasattr(socket, "AF_UNIX"):
+        # The daemon only listens on AF_UNIX, so a platform without it cannot
+        # run the accelerator at all. Windows is the one case where a missing
+        # ``socket.AF_UNIX`` says nothing about the platform — the OS and
+        # gRPC's C-core both support the family, only CPython's socket module
+        # does not — so it is excluded from the check.
+        if not hasattr(socket, "AF_UNIX") and not _IS_WINDOWS:
             raise OSError(
                 "Unix domain sockets (AF_UNIX) are not supported on this platform."
             )
 
+        # mkdtemp is what confines the socket to this user: 0700 on POSIX, and
+        # on Windows an inherited ACL from the per-user %TEMP% tree. The latter
+        # is the only access control the socket gets there — Windows has no
+        # umask and chmod only toggles the read-only bit — so the daemon
+        # deliberately relies on the directory we pick here.
         self._tempdir = tempfile.mkdtemp(prefix=_TEMPDIR_PREFIX)
         self._uds_path = os.path.join(self._tempdir, _SOCKET_FILENAME)
 
@@ -346,7 +385,12 @@ def close(self) -> None:
                     except OSError:
                         pass
                 if not self._wait_for_exit(_STDIN_GRACE_SECONDS):
-                    # Step 2: SIGTERM.
+                    # Step 2: SIGTERM. On Windows this is TerminateProcess, so
+                    # steps 2 and 3 are the same hard kill and the daemon's one
+                    # chance to shut down cleanly is the stdin EOF above —
+                    # which is also the signal it uses there to notice the
+                    # parent died, since a ppid watchdog cannot work on
+                    # Windows.
                     try:
                         proc.terminate()
                     except OSError:
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py
index 7b9b94993535..e6953f676664 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py
@@ -46,6 +46,18 @@
 _READ_ROWS_METHOD = "/google.bigtable.v2.Bigtable/ReadRows"
 
 
+def _uds_target(uds_path: str) -> str:
+    """Build the gRPC target for the daemon's socket at ``uds_path``.
+
+    Must be the single-colon form. ``unix://`` opens an *authority* component,
+    which gRPC only accepts when the rest is an absolute POSIX path; hand it a
+    Windows path and ``C:`` parses as the authority, so channel construction
+    fails outright with "authority-based URIs not supported by the unix
+    scheme". ``unix:`` takes the remainder as a plain path on every platform.
+    """
+    return f"unix:{uds_path}"
+
+
 @CrossSync.convert_class(sync_name="_AcceleratorClient")
 class _AsyncAcceleratorClient:
     """Thin gRPC client bound to the daemon's UDS.
@@ -58,7 +70,7 @@ class _AsyncAcceleratorClient:
     def __init__(self, uds_path: str, auth_secret: str):
         self._uds_path = uds_path
         self._metadata = (("x-accelerator-token", auth_secret),)
-        self._channel = insecure_channel(f"unix://{uds_path}")
+        self._channel = insecure_channel(_uds_target(uds_path))
         self._mutate_row_stub = self._channel.unary_unary(
             _MUTATE_ROW_METHOD,
             request_serializer=MutateRowRequest.serialize,
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py
index ea9168abae6b..73c866f24553 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py
@@ -20,7 +20,6 @@
 import logging
 import os
 import random
-import sys
 import time
 import warnings
 import weakref
@@ -1384,13 +1383,6 @@ def _maybe_start_accelerator(self, *, explicit: bool) -> None:
         hard error rather than a silent fallback, both for the emulator (a
         genuine misconfiguration) and for a daemon that fails to start.
         """
-        if sys.platform == "win32":
-            if explicit:
-                raise RuntimeError(
-                    "use_accelerator=True is not supported on Windows; pass "
-                    "use_accelerator=False to use the native client."
-                )
-            return
         if self.client._emulator_host is not None:
             if explicit:
                 raise RuntimeError(
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py
index bd0965843620..51905d4b04cd 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py
@@ -43,6 +43,17 @@
 _READ_ROWS_METHOD = "/google.bigtable.v2.Bigtable/ReadRows"
 
 
+def _uds_target(uds_path: str) -> str:
+    """Build the gRPC target for the daemon's socket at ``uds_path``.
+
+    Must be the single-colon form. ``unix://`` opens an *authority* component,
+    which gRPC only accepts when the rest is an absolute POSIX path; hand it a
+    Windows path and ``C:`` parses as the authority, so channel construction
+    fails outright with "authority-based URIs not supported by the unix
+    scheme". ``unix:`` takes the remainder as a plain path on every platform."""
+    return f"unix:{uds_path}"
+
+
 class _AcceleratorClient:
     """Thin gRPC client bound to the daemon's UDS.
 
@@ -54,7 +65,7 @@ class _AcceleratorClient:
     def __init__(self, uds_path: str, auth_secret: str):
         self._uds_path = uds_path
         self._metadata = (("x-accelerator-token", auth_secret),)
-        self._channel = insecure_channel(f"unix://{uds_path}")
+        self._channel = insecure_channel(_uds_target(uds_path))
         self._mutate_row_stub = self._channel.unary_unary(
             _MUTATE_ROW_METHOD,
             request_serializer=MutateRowRequest.serialize,
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py
index 67a8b3f12175..9f767ec87ca6 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py
@@ -23,7 +23,6 @@
 import logging
 import os
 import random
-import sys
 import time
 import warnings
 import weakref
@@ -1079,12 +1078,6 @@ def _maybe_start_accelerator(self, *, explicit: bool) -> None:
         ``use_accelerator=True``: a caller who required the accelerator gets a
         hard error rather than a silent fallback, both for the emulator (a
         genuine misconfiguration) and for a daemon that fails to start."""
-        if sys.platform == "win32":
-            if explicit:
-                raise RuntimeError(
-                    "use_accelerator=True is not supported on Windows; pass use_accelerator=False to use the native client."
-                )
-            return
         if self.client._emulator_host is not None:
             if explicit:
                 raise RuntimeError(
diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py
index 458f80857ffa..727628bb3c4d 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py
@@ -99,7 +99,9 @@ def __init__(self, uds_path, *, status=ServingStatus.SERVING, serve_health=True)
             self._server.add_generic_rpc_handlers(
                 (grpc.method_handlers_generic_handler(service, {method: handler}),)
             )
-        self._server.add_insecure_port(f"unix://{uds_path}")
+        # Single-colon form, same as the client — ``unix://`` cannot express a
+        # Windows path. See ``_accelerator_client._uds_target``.
+        self._server.add_insecure_port(f"unix:{uds_path}")
         self._server.start()
 
     def _check(self, request, context):
diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py
index e2619cb9d523..543e31c88b8b 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py
@@ -89,7 +89,7 @@ def __init__(self, uds_path, *, status=ServingStatus.SERVING, serve_health=True)
             self._server.add_generic_rpc_handlers(
                 (grpc.method_handlers_generic_handler(service, {method: handler}),)
             )
-        self._server.add_insecure_port(f"unix://{uds_path}")
+        self._server.add_insecure_port(f"unix:{uds_path}")
         self._server.start()
 
     def _check(self, request, context):
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
index 2c3fb6a0108e..a118e4f37b96 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
@@ -205,6 +205,17 @@ def fake_mkdtemp(prefix=""):
     yield
 
 
+@pytest.fixture
+def no_af_unix(monkeypatch):
+    """Make the socket module look the way CPython's does on Windows.
+
+    The family is simply absent there — the OS and gRPC's C-core support
+    AF_UNIX, the socket module just never exposed it — so every branch keyed on
+    ``hasattr(socket, "AF_UNIX")`` is reachable from a POSIX test run.
+    """
+    monkeypatch.delattr(_daemon.socket, "AF_UNIX", raising=False)
+
+
 # ---------------------------------------------------------------------------
 # Properties before/after start
 # ---------------------------------------------------------------------------
@@ -258,6 +269,17 @@ def test_missing_binary_raises_file_not_found(self, tmp_path):
             with pytest.raises(FileNotFoundError, match="Accelerator binary not found"):
                 _resolve_binary_path()
 
+    @pytest.mark.parametrize(
+        "is_windows,expected",
+        [(False, "bin/accelerator"), (True, "bin/accelerator.exe")],
+    )
+    def test_binary_name_follows_platform(self, monkeypatch, is_windows, expected):
+        """Windows wheels stage the binary under the .exe name Go gives it."""
+        monkeypatch.setattr(_daemon, "_IS_WINDOWS", is_windows)
+        with unittest.mock.patch("os.path.isfile", return_value=True):
+            resolved = _daemon._resolve_binary_path()
+        assert resolved.endswith(expected)
+
 
 # ---------------------------------------------------------------------------
 # start() failure modes
@@ -315,23 +337,10 @@ def test_uds_never_appears_raises_timeout(self, tmp_path, monkeypatch):
         with pytest.raises(RuntimeError, match="did not become ready"):
             d.start()
 
-    def test_no_af_unix_raises_os_error(self, tmp_path, monkeypatch):
-        import socket as _socket
-
-        monkeypatch.delattr(_socket, "AF_UNIX", raising=False)
-        # Reload the attribute check in _daemon
-        monkeypatch.setattr(_daemon.socket, "AF_UNIX", None, raising=False)
-        # Simpler: just check that hasattr(socket, "AF_UNIX") guards the call.
-        with (
-            unittest.mock.patch.object(_daemon.socket, "AF_UNIX", None, create=True),
-            unittest.mock.patch(
-                "google.cloud.bigtable.data._accelerator._daemon.hasattr",
-                return_value=False,
-            ),
-        ):
-            d = _make_daemon(tmp_path)
-            with pytest.raises(OSError, match="AF_UNIX"):
-                d.start()
+    def test_no_af_unix_raises_os_error(self, tmp_path, monkeypatch, no_af_unix):
+        d = _make_daemon(tmp_path)
+        with pytest.raises(OSError, match="AF_UNIX"):
+            d.start()
 
 
 # ---------------------------------------------------------------------------
@@ -728,3 +737,69 @@ def test_false_for_socket_file_whose_owner_exited(self, tmp_path):
         sock.close()
         assert os.path.exists(path)
         assert _daemon._socket_is_live(path) is False
+
+
+# ---------------------------------------------------------------------------
+# Windows (win_amd64 / win32)
+# ---------------------------------------------------------------------------
+
+
+class TestWindows:
+    """Windows reaches the daemon over the same AF_UNIX socket as every other
+    platform. What differs is that CPython's socket module does not expose the
+    family, so anything the other platforms answer with a probe connection has
+    to be read off the filesystem instead."""
+
+    def test_start_is_not_blocked_by_missing_af_unix(
+        self, tmp_path, monkeypatch, no_af_unix
+    ):
+        """The AF_UNIX guard must not fire on Windows: a missing family there
+        says something about CPython, not about the platform."""
+        monkeypatch.setattr(_daemon, "_IS_WINDOWS", True)
+
+        def fake_popen(argv, **kwargs):
+            raise _StopStart()
+
+        monkeypatch.setattr(_daemon.subprocess, "Popen", fake_popen)
+        d = _make_daemon(tmp_path)
+        # Reaching Popen means the guard let us through.
+        with pytest.raises(_StopStart):
+            d.start()
+
+    def test_socket_is_live_falls_back_to_path_existence(self, tmp_path, no_af_unix):
+        sock = tmp_path / _daemon._SOCKET_FILENAME
+        assert _daemon._socket_is_live(str(sock)) is False
+        sock.touch()
+        assert _daemon._socket_is_live(str(sock)) is True
+
+    def test_uds_target_accepts_a_windows_path(self):
+        """The channel target must survive a drive-lettered path.
+
+        ``unix://`` opens a URI authority component, so ``unix://C:\\...``
+        parses ``C:`` as the authority and gRPC refuses to build the channel at
+        all — the accelerator would start, verify, and then fail every routed
+        RPC. Constructing the channel is the assertion; it raises on a bad
+        target before any connection is attempted.
+        """
+        import grpc
+
+        from google.cloud.bigtable.data._sync_autogen._accelerator_client import (
+            _uds_target,
+        )
+
+        win_path = r"C:\Users\me\AppData\Local\Temp\bt-accel-abc123\sock"
+        assert _uds_target(win_path) == "unix:" + win_path
+        channel = grpc.insecure_channel(_uds_target(win_path))
+        try:
+            stub = channel.unary_unary(
+                "/google.bigtable.v2.Bigtable/MutateRow",
+                request_serializer=lambda _: b"",
+                response_deserializer=lambda payload: payload,
+            )
+            with pytest.raises(grpc.RpcError) as exc_info:
+                stub(None, timeout=0.5)
+        finally:
+            channel.close()
+        # UNAVAILABLE means the target resolved and nothing was listening, which
+        # is the expected outcome. A malformed target fails earlier, as UNKNOWN.
+        assert exc_info.value.code() == grpc.StatusCode.UNAVAILABLE

From ff4f1d37d3e3732f7bc1091cd323ca233d40b10f Mon Sep 17 00:00:00 2001
From: Mattie Fu <mattiefu@google.com>
Date: Tue, 20 Oct 2026 17:56:17 +0000
Subject: [PATCH 2/3] test(bigtable): shorten the health test UDS path and
 cover the async _uds_target

Change-Id: I4ee8c69a938ff86ab6967642809a70f5330af25b
---
 .../unit/data/_async/test__accelerator_health.py  | 15 +++++++++++++++
 .../_sync_autogen/test__accelerator_health.py     |  9 +++++++++
 .../tests/unit/data/test_accelerator_daemon.py    |  4 ++++
 3 files changed, 28 insertions(+)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py
index 727628bb3c4d..c7e8a9a1a3c0 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py
@@ -26,6 +26,8 @@
 ``tests/unit/data/test_accelerator_health.py``.
 """
 
+import pathlib
+import tempfile
 import threading
 import time
 from concurrent import futures
@@ -68,6 +70,19 @@
 }
 
 
+# sockaddr_un.sun_path is 108 bytes on Linux and Windows (104 on macOS).
+# Pytest's built-in tmp_path nests pytest-of-<user>/pytest-N/<test>/ under
+# %TEMP%, which on Windows and macOS can exceed 107 chars and make
+# add_insecure_port fail to bind. Matching the daemon's own
+# tempfile.mkdtemp(prefix="bt-accel-") placement keeps the socket path well
+# under the limit on every platform.
+@pytest.fixture
+def tmp_path():
+    """Short UDS directory directly under the system temp root."""
+    with tempfile.TemporaryDirectory(prefix="bt-accel-") as d:
+        yield pathlib.Path(d)
+
+
 class _FakeDaemonHealthServer:
     """A real gRPC server standing in for the daemon's health service.
 
diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py
index 543e31c88b8b..e9508d0761be 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py
@@ -29,6 +29,8 @@
 ``tests/unit/data/test_accelerator_health.py``.
 """
 
+import pathlib
+import tempfile
 import threading
 import time
 from concurrent import futures
@@ -58,6 +60,13 @@
 }
 
 
+@pytest.fixture
+def tmp_path():
+    """Short UDS directory directly under the system temp root."""
+    with tempfile.TemporaryDirectory(prefix="bt-accel-") as d:
+        yield pathlib.Path(d)
+
+
 class _FakeDaemonHealthServer:
     """A real gRPC server standing in for the daemon's health service.
 
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
index a118e4f37b96..efc7aa847af2 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
@@ -783,12 +783,16 @@ def test_uds_target_accepts_a_windows_path(self):
         """
         import grpc
 
+        from google.cloud.bigtable.data._async._accelerator_client import (
+            _uds_target as _async_uds_target,
+        )
         from google.cloud.bigtable.data._sync_autogen._accelerator_client import (
             _uds_target,
         )
 
         win_path = r"C:\Users\me\AppData\Local\Temp\bt-accel-abc123\sock"
         assert _uds_target(win_path) == "unix:" + win_path
+        assert _async_uds_target(win_path) == _uds_target(win_path)
         channel = grpc.insecure_channel(_uds_target(win_path))
         try:
             stub = channel.unary_unary(

From fc1e22a01dce07c2a1784fc1176f606b7bb5dbaa Mon Sep 17 00:00:00 2001
From: Mattie Fu <mattiefu@google.com>
Date: Thu, 22 Oct 2026 16:36:35 +0000
Subject: [PATCH 3/3] feat(bigtable): probe daemon readiness with a gRPC
 channel instead of a raw socket connect

Change-Id: I1f9899ed28dc504fbdbfdb2a93d34a404e187301
---
 .../bigtable/data/_accelerator/_daemon.py     | 112 +++++++++---------
 .../data/_async/_accelerator_client.py        |  13 +-
 .../data/_sync_autogen/_accelerator_client.py |  12 +-
 .../unit/data/test_accelerator_daemon.py      |  81 +++++++------
 4 files changed, 103 insertions(+), 115 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py
index e3fbd9cb24ed..887d43446943 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py
@@ -16,9 +16,11 @@
 
 The daemon binary embeds an in-process Go Bigtable client and exposes the
 ``google.bigtable.v2.Bigtable`` service over a Unix domain socket. This module
-owns spawning the binary, waiting for the UDS to become connectable, and
-tearing the process down. It does NOT speak gRPC; that's the
-``_AcceleratorClient`` companion's job.
+owns spawning the binary, waiting for its gRPC server to come up, and tearing
+the process down. It issues no RPCs — that's the ``_AcceleratorClient``
+companion's job — but it does open a throwaway channel to decide when the
+daemon is ready, since on Windows that is the only probe available from
+Python (see ``_wait_until_ready``).
 """
 
 from __future__ import annotations
@@ -35,11 +37,13 @@
 import time
 from typing import Any, Mapping, Sequence
 
+import grpc
+
 # Windows differs from every other supported platform in two ways that matter
 # here, so it gets a named constant rather than scattered ``sys.platform``
 # checks: the daemon binary carries a ``.exe`` suffix, and CPython's socket
 # module exposes no ``AF_UNIX`` even though the OS and gRPC's C-core both speak
-# it (see ``_socket_is_live``).
+# it (see ``start`` and ``_wait_until_ready``).
 _IS_WINDOWS = sys.platform == "win32"
 
 # Environment variable that overrides the daemon startup readiness timeout, in
@@ -87,17 +91,15 @@
 # ``_resolve_startup_timeout``).
 _DEFAULT_STARTUP_TIMEOUT = 20.0
 
+# How long each poll of the readiness future waits before the loop re-checks
+# that the daemon process is still alive. Doubles as the loop's sleep.
+_READY_POLL_SECONDS = 0.05
+
 # Sequence: close stdin, wait this long; SIGTERM, wait again; SIGKILL.
 _STDIN_GRACE_SECONDS = 2.0
 _SIGTERM_GRACE_SECONDS = 2.0
 _SIGKILL_GRACE_SECONDS = 2.0
 
-# How long a single connect() probe of a daemon's socket may take. A UDS
-# connect against a listening socket completes in the kernel without the
-# daemon doing anything, so this only ever elapses on a badly overloaded
-# host — where the honest answer is "I cannot tell", not "it is dead".
-_SOCKET_PROBE_TIMEOUT_SECONDS = 0.25
-
 
 def _resolve_binary_path() -> str:
     """Return the path of the bundled daemon binary, raising if it is absent."""
@@ -138,38 +140,16 @@ def _resolve_startup_timeout() -> float:
     return value
 
 
-def _socket_is_live(uds_path: str) -> bool:
-    """Return True if something is currently accepting connections on the UDS.
+def _uds_target(uds_path: str) -> str:
+    """Build the gRPC target for the daemon's socket at ``uds_path``.
 
-    Used by ``_wait_until_ready`` to poll for the daemon coming up, so every
-    failure simply answers "not yet" and the caller tries again.
+    Must be the single-colon form. ``unix://`` opens an *authority* component,
+    which gRPC only accepts when the rest is an absolute POSIX path; hand it a
+    Windows path and ``C:`` parses as the authority, so channel construction
+    fails outright with "authority-based URIs not supported by the unix
+    scheme". ``unix:`` takes the remainder as a plain path on every platform.
     """
-    if not os.path.exists(uds_path):
-        return False
-    if not hasattr(socket, "AF_UNIX"):
-        # Windows. The OS (10 1803+ / Server 2019+) and gRPC's C-core both
-        # support AF_UNIX, which is why the accelerator runs here at all, but
-        # CPython's socket module still does not expose the family, so we
-        # cannot open a probe connection from Python. The gate is
-        # ``HAVE_SYS_UN_H``: ``PC/pyconfig.h`` leaves it commented out and
-        # ``Modules/socketmodule.h`` responds with ``#undef AF_UNIX``, so the
-        # attribute is absent on every current CPython for Windows. Testing
-        # the attribute rather than the platform keeps this correct for any
-        # other build that lacks the family, and is what lets the tests
-        # exercise this branch from a POSIX run.
-        #
-        # Fall back to the bind artifact: the daemon's
-        # ``net.Listen("unix", ...)`` creates this file and puts the socket in
-        # the listen backlog inside one call, so the path existing means the
-        # listener is up.
-        return True
-    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
-        probe.settimeout(_SOCKET_PROBE_TIMEOUT_SECONDS)
-        try:
-            probe.connect(uds_path)
-            return True
-        except OSError:
-            return False
+    return f"unix:{uds_path}"
 
 
 class AcceleratorDaemon:
@@ -440,25 +420,51 @@ def read_identity(self) -> dict[str, Any]:
                 pass
 
     def _wait_until_ready(self, timeout: float) -> None:
-        """Poll until the UDS accepts a connection, the child dies, or timeout.
+        """Poll until the daemon's gRPC server is reachable, the child dies,
+        or timeout.
+
+        Readiness is a gRPC channel reaching ``READY``, not a bare socket
+        connect. Two reasons. It is the stronger signal: ``net.Listen`` puts
+        the socket in the listen backlog, so a raw connect succeeds before the
+        Go gRPC server has accepted anything, whereas ``READY`` means the
+        HTTP/2 handshake completed and an RPC would actually be served. And it
+        is the only signal available on Windows, where CPython's socket module
+        has no ``AF_UNIX`` (``PC/pyconfig.h`` leaves ``HAVE_SYS_UN_H``
+        commented out) even though the OS and gRPC's C-core both support the
+        family — which is why the accelerator runs there at all.
 
         Raises RuntimeError if the daemon exits during startup or does not
-        become connectable within ``timeout`` seconds.
+        become ready within ``timeout`` seconds.
         """
         if self._proc is None or self._uds_path is None:
             raise RuntimeError("AcceleratorDaemon._wait_until_ready() before spawn")
         deadline = time.monotonic() + timeout
-        while time.monotonic() < deadline:
-            exit_code = self._proc.poll()
-            if exit_code is not None:
-                log_tail = self._read_log_tail()
-                raise RuntimeError(
-                    "Accelerator daemon exited during startup "
-                    f"(exit code {exit_code}). log: {log_tail!r}"
-                )
-            if _socket_is_live(self._uds_path):
-                return
-            time.sleep(0.05)
+        # One channel for the whole wait. gRPC connects lazily and retries
+        # with backoff on its own, so the future below is re-polled rather
+        # than rebuilt; each short wait on it doubles as this loop's sleep.
+        channel = grpc.insecure_channel(_uds_target(self._uds_path))
+        ready = grpc.channel_ready_future(channel)
+        try:
+            while time.monotonic() < deadline:
+                exit_code = self._proc.poll()
+                if exit_code is not None:
+                    log_tail = self._read_log_tail()
+                    raise RuntimeError(
+                        "Accelerator daemon exited during startup "
+                        f"(exit code {exit_code}). log: {log_tail!r}"
+                    )
+                try:
+                    ready.result(timeout=_READY_POLL_SECONDS)
+                    return
+                except grpc.FutureTimeoutError:
+                    # Not up yet. Nothing distinguishes "still starting" from
+                    # "never will" here — a failed connect leaves the channel
+                    # in TRANSIENT_FAILURE and it keeps retrying — so the
+                    # deadline above is what ends the wait.
+                    continue
+        finally:
+            ready.cancel()
+            channel.close()
         log_tail = self._read_log_tail()
         raise RuntimeError(
             "Accelerator daemon did not become ready within "
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py
index e6953f676664..e6c2dd9db41c 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py
@@ -21,6 +21,7 @@
 
 from __future__ import annotations
 
+from google.cloud.bigtable.data._accelerator._daemon import _uds_target
 from google.cloud.bigtable.data._accelerator._health import (
     HEALTH_CHECK_METHOD,
     ServingStatus,
@@ -46,18 +47,6 @@
 _READ_ROWS_METHOD = "/google.bigtable.v2.Bigtable/ReadRows"
 
 
-def _uds_target(uds_path: str) -> str:
-    """Build the gRPC target for the daemon's socket at ``uds_path``.
-
-    Must be the single-colon form. ``unix://`` opens an *authority* component,
-    which gRPC only accepts when the rest is an absolute POSIX path; hand it a
-    Windows path and ``C:`` parses as the authority, so channel construction
-    fails outright with "authority-based URIs not supported by the unix
-    scheme". ``unix:`` takes the remainder as a plain path on every platform.
-    """
-    return f"unix:{uds_path}"
-
-
 @CrossSync.convert_class(sync_name="_AcceleratorClient")
 class _AsyncAcceleratorClient:
     """Thin gRPC client bound to the daemon's UDS.
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py
index 51905d4b04cd..452fcc9c7b72 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py
@@ -26,6 +26,7 @@
 
 from grpc import insecure_channel
 
+from google.cloud.bigtable.data._accelerator._daemon import _uds_target
 from google.cloud.bigtable.data._accelerator._health import (
     HEALTH_CHECK_METHOD,
     ServingStatus,
@@ -43,17 +44,6 @@
 _READ_ROWS_METHOD = "/google.bigtable.v2.Bigtable/ReadRows"
 
 
-def _uds_target(uds_path: str) -> str:
-    """Build the gRPC target for the daemon's socket at ``uds_path``.
-
-    Must be the single-colon form. ``unix://`` opens an *authority* component,
-    which gRPC only accepts when the rest is an absolute POSIX path; hand it a
-    Windows path and ``C:`` parses as the authority, so channel construction
-    fails outright with "authority-based URIs not supported by the unix
-    scheme". ``unix:`` takes the remainder as a plain path on every platform."""
-    return f"unix:{uds_path}"
-
-
 class _AcceleratorClient:
     """Thin gRPC client bound to the daemon's UDS.
 
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
index efc7aa847af2..c8f4c8df638a 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py
@@ -15,7 +15,6 @@
 import json
 import os
 import re
-import socket
 import unittest.mock
 
 import pytest
@@ -330,8 +329,8 @@ def test_uds_never_appears_raises_timeout(self, tmp_path, monkeypatch):
         proc_mock.stdin = unittest.mock.Mock()
 
         monkeypatch.setattr(_daemon.subprocess, "Popen", lambda *a, **kw: proc_mock)
-        # Make os.path.exists always False so UDS never appears.
-        monkeypatch.setattr(_daemon.os.path, "exists", lambda p: False)
+        # Nothing ever binds the socket, so the readiness channel stays in
+        # TRANSIENT_FAILURE and the deadline is what ends the wait.
         monkeypatch.setenv(_STARTUP_TIMEOUT_ENV_VAR, "0.05")
         d = _make_daemon(tmp_path)
         with pytest.raises(RuntimeError, match="did not become ready"):
@@ -709,36 +708,6 @@ def test_daemon_constructor_invalid_env_var_raises(self, tmp_path, monkeypatch):
             _make_daemon(tmp_path)
 
 
-@pytest.mark.skipif(
-    not hasattr(_daemon.socket, "AF_UNIX"),
-    reason="Unix domain sockets (AF_UNIX) are not supported on this platform",
-)
-@pytest.mark.skipif(not hasattr(_daemon.socket, "AF_UNIX"), reason="requires AF_UNIX")
-class TestSocketIsLive:
-    """_socket_is_live backs the _wait_until_ready poll: is the daemon up?"""
-
-    def test_true_while_listening(self, tmp_path):
-        path = str(tmp_path / "sock")
-        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
-            sock.bind(path)
-            sock.listen(1)
-            assert _daemon._socket_is_live(path) is True
-
-    def test_false_when_path_is_absent(self, tmp_path):
-        assert _daemon._socket_is_live(str(tmp_path / "nope")) is False
-
-    def test_false_for_socket_file_whose_owner_exited(self, tmp_path):
-        # The socket file outlives the process that bound it; nothing is
-        # accepting, so the daemon is not up yet (or any more).
-        path = str(tmp_path / "sock")
-        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        sock.bind(path)
-        sock.listen(1)
-        sock.close()
-        assert os.path.exists(path)
-        assert _daemon._socket_is_live(path) is False
-
-
 # ---------------------------------------------------------------------------
 # Windows (win_amd64 / win32)
 # ---------------------------------------------------------------------------
@@ -747,8 +716,8 @@ def test_false_for_socket_file_whose_owner_exited(self, tmp_path):
 class TestWindows:
     """Windows reaches the daemon over the same AF_UNIX socket as every other
     platform. What differs is that CPython's socket module does not expose the
-    family, so anything the other platforms answer with a probe connection has
-    to be read off the filesystem instead."""
+    family, so nothing here may depend on ``socket.AF_UNIX`` — readiness goes
+    through gRPC's C-core, which does speak it."""
 
     def test_start_is_not_blocked_by_missing_af_unix(
         self, tmp_path, monkeypatch, no_af_unix
@@ -766,11 +735,45 @@ def fake_popen(argv, **kwargs):
         with pytest.raises(_StopStart):
             d.start()
 
-    def test_socket_is_live_falls_back_to_path_existence(self, tmp_path, no_af_unix):
+    def test_readiness_does_not_touch_socket_af_unix(
+        self, tmp_path, monkeypatch, no_af_unix
+    ):
+        """Readiness must work with no ``socket.AF_UNIX`` at all.
+
+        This is the whole reason the probe is a gRPC channel: with the family
+        absent, the wait still has to reach its timeout honestly rather than
+        raise AttributeError or shortcut to "ready" off the bind artifact.
+        """
+        import subprocess
+
+        proc_mock = unittest.mock.Mock(spec=subprocess.Popen)
+        proc_mock.poll.return_value = None  # alive, but never binds
+        proc_mock.stdin = unittest.mock.Mock()
+        monkeypatch.setattr(_daemon, "_IS_WINDOWS", True)
+        monkeypatch.setattr(_daemon.subprocess, "Popen", lambda *a, **kw: proc_mock)
+        monkeypatch.setenv(_STARTUP_TIMEOUT_ENV_VAR, "0.05")
+        d = _make_daemon(tmp_path)
+        with pytest.raises(RuntimeError, match="did not become ready"):
+            d.start()
+
+    def test_readiness_reaches_a_real_grpc_server(self, tmp_path):
+        """And the positive case: a server that completes the HTTP/2 handshake
+        is what READY means, so the probe returns rather than timing out."""
+        from concurrent import futures
+
+        import grpc
+
         sock = tmp_path / _daemon._SOCKET_FILENAME
-        assert _daemon._socket_is_live(str(sock)) is False
-        sock.touch()
-        assert _daemon._socket_is_live(str(sock)) is True
+        server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
+        server.add_insecure_port(_daemon._uds_target(str(sock)))
+        server.start()
+        try:
+            d = _make_daemon(tmp_path)
+            d._proc = unittest.mock.Mock(**{"poll.return_value": None})
+            d._uds_path = str(sock)
+            d._wait_until_ready(10.0)  # returns, or the test times out
+        finally:
+            server.stop(None)
 
     def test_uds_target_accepts_a_windows_path(self):
         """The channel target must survive a drive-lettered path.