fix(sphinxdocs): materialize source symlinks to resolve relative cross-references (#3977)

sphinxdocs/sphinxdocs/private/BUILD.bazel

@@ -129,6 +129,12 @@ py_binary(
     deps = [":proto_to_markdown_lib"],
 )
 
+py_library(
+    name = "sphinx_build_lib",
+    srcs = ["sphinx_build.py"],
+    visibility = ["//:__subpackages__"],
+)
+
 py_library(
     name = "proto_to_markdown_lib",
     srcs = ["proto_to_markdown.py"],

sphinxdocs/sphinxdocs/private/sphinx.bzl

@@ -360,12 +360,17 @@ def _sphinx_source_tree_impl(ctx):
         dest_path = paths.join(source_prefix, dest_path)
         if source_file.is_directory:
             dest_file = ctx.actions.declare_directory(dest_path)
+            progress_message = "Symlinking Sphinx source directory %{input} to %{output}"
         else:
             dest_file = ctx.actions.declare_file(dest_path)
+            progress_message = "Symlinking Sphinx source %{input} to %{output}"
+
+        # NOTE: Sphinx/MyST will read through symlinks, which can break relative
+        # xref lookup. Files are copied during the action phase to prevent this.
         ctx.actions.symlink(
             output = dest_file,
             target_file = source_file,
-            progress_message = "Symlinking Sphinx source %{input} to %{output}",
+            progress_message = progress_message,
         )
         sphinx_source_files.append(dest_file)
         return dest_file

sphinxdocs/sphinxdocs/private/sphinx_build.py

@@ -1,10 +1,14 @@
+import concurrent.futures
 import contextlib
 import io
 import json
 import logging
 import os
+import pathlib
 import shutil
+import stat
 import sys
+import threading
 import traceback
 import typing
 
@@ -29,7 +33,171 @@ def __init__(self, message, exit_code):
 _REQUEST_INFO_CONFIG_NAME = "bazel_worker_request_info_path"
 
 
+class DirectorySyncerError(Exception):
+    """Raised when one or more errors occur during directory synchronization."""
+
+    def __init__(self, errors: typing.List[BaseException]):
+        self.errors = errors
+        message = f"Encountered {len(errors)} error(s) during sync:\n" + "\n".join(
+            f"  - {e}" for e in errors
+        )
+        super().__init__(message)
+
+
+class DirectorySyncer:
+    """Synchronizes a working destination directory from a source directory.
+
+    Supports concurrent SHA-aware incremental updates (via sync()) for worker
+    mode and concurrent full directory copying (via copytree()) for non-worker
+    mode, ensuring physical file materialization to prevent relative
+    cross-reference resolution failures.
+    """
+
+    def __init__(
+        self,
+        srcdir: pathlib.Path,
+        destdir: pathlib.Path,
+        max_workers: typing.Optional[int] = None,
+    ):
+        self._srcdir = srcdir
+        self._destdir = destdir
+        self._max_workers = max_workers or min(32, (os.cpu_count() or 4) + 4)
+        self._current_shas: typing.Dict[str, str] = {}
+        self._lock = threading.Lock()
+        self._finished_cond = threading.Condition(self._lock)
+        self._remaining = 0
+        self._errors: typing.List[BaseException] = []
+        self._executor: typing.Optional[concurrent.futures.ThreadPoolExecutor] = None
+
+    def _reset_state(self) -> None:
+        with self._lock:
+            self._errors.clear()
+            self._remaining = 0
+
+    def _wait_for_completion(self) -> None:
+        with self._lock:
+            while self._remaining > 0:
+                self._finished_cond.wait()
+        if self._errors:
+            raise DirectorySyncerError(list(self._errors))
+
+    def _submit_task(self, fn, *args) -> None:
+        with self._lock:
+            self._remaining += 1
+        future = self._executor.submit(fn, *args)
+        future.add_done_callback(self._handle_task_done)
+
+    def _handle_task_done(self, future: concurrent.futures.Future) -> None:
+        exc = future.exception()
+        if exc:
+            with self._lock:
+                self._errors.append(exc)
+
+    def _task_finished(self) -> None:
+        with self._lock:
+            self._remaining -= 1
+            if self._remaining == 0:
+                self._finished_cond.notify_all()
+
+    @contextlib.contextmanager
+    def _create_executor(self):
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=self._max_workers
+        ) as executor:
+            self._executor = executor
+            try:
+                yield
+            finally:
+                self._executor = None
+
+    def copytree(self) -> None:
+        """Concurrently copies srcdir to destdir without SHA tracking."""
+        self._reset_state()
+        shutil.rmtree(self._destdir, ignore_errors=True)
+        with self._create_executor():
+            self._submit_task(self._copy_dir, self._srcdir, self._destdir)
+            self._wait_for_completion()
+
+    def sync(self, entries: typing.Dict[str, str]) -> None:
+        """Synchronizes destdir to match entries {relative_path: sha} concurrently."""
+        self._reset_state()
+
+        to_remove = set(self._current_shas.keys()) - set(entries.keys())
+        to_copy = {
+            path: sha
+            for path, sha in entries.items()
+            if self._current_shas.get(path) != sha
+        }
+
+        if not to_remove and not to_copy:
+            self._current_shas = dict(entries)
+            return
+
+        with self._create_executor():
+            # 1. Submit stale path removals concurrently ASAP
+            for rel_path in to_remove:
+                dest_path = self._destdir / rel_path
+                self._submit_task(self._remove_path, dest_path)
+
+            # 2. Submit created/updated item copies concurrently ASAP
+            for rel_path in to_copy:
+                src_path = self._srcdir / rel_path
+                dest_path = self._destdir / rel_path
+                if src_path.is_dir():
+                    self._submit_task(self._copy_dir, src_path, dest_path)
+                else:
+                    self._submit_task(self._copy_file, src_path, dest_path)
+
+            self._wait_for_completion()
+
+        self._current_shas = dict(entries)
+
+    def _remove_path(self, dest_path: pathlib.Path) -> None:
+        try:
+            if dest_path.is_dir() and not dest_path.is_symlink():
+                shutil.rmtree(dest_path)
+            else:
+                dest_path.unlink(missing_ok=True)
+        except BaseException as e:
+            e.add_note(f"Failed removing path: dest_path={dest_path}")
+            raise
+        finally:
+            self._task_finished()
+
+    def _copy_file(self, src: pathlib.Path, dest: pathlib.Path) -> None:
+        try:
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(src, dest)
+            mode = dest.stat().st_mode
+            dest.chmod(mode | stat.S_IWUSR)
+        except BaseException as e:
+            e.add_note(f"Failed copying file: src={src}, dest={dest}")
+            raise
+        finally:
+            self._task_finished()
+
+    def _copy_dir(self, src: pathlib.Path, dest: pathlib.Path) -> None:
+        """Recursively creates destination directory and submits tasks for its entries."""
+        try:
+            dest.mkdir(parents=True, exist_ok=True)
+            with os.scandir(src) as scanner:
+                for entry in scanner:
+                    c_dest = dest / entry.name
+                    c_src = pathlib.Path(entry.path)
+                    if entry.is_dir():
+                        self._submit_task(self._copy_dir, c_src, c_dest)
+                    else:
+                        self._submit_task(self._copy_file, c_src, c_dest)
+        except BaseException as e:
+            e.add_note(f"Failed copying directory: src={src}, dest={dest}")
+            raise
+        finally:
+            self._task_finished()
+
+
 class Worker:
+    """A Bazel persistent worker for Sphinx builds."""
+
     def __init__(
         self, instream: "typing.TextIO", outstream: "typing.TextIO", exec_root: str
     ):
@@ -51,6 +219,7 @@ def __init__(
 
         # dict[str srcdir, dict[str path, str digest]]
         self._digests = {}
+        self._syncers: typing.Dict[pathlib.Path, DirectorySyncer] = {}
 
         # Internal output directories the worker gives to Sphinx that need
         # to be cleaned up upon exit.
@@ -124,18 +293,25 @@ def _send_response(self, response: "WorkResponse") -> None:
 
     def _prepare_sphinx(self, request):
         sphinx_args = request["arguments"]
-        srcdir = sphinx_args[0]
+        srcdir = pathlib.Path(sphinx_args[0])
+        destdir = pathlib.Path(f"{srcdir}.worker-in.d")
 
         incoming_digests = {}
-        current_digests = self._digests.setdefault(srcdir, {})
+        current_digests = self._digests.setdefault(str(srcdir), {})
         is_first_request = not current_digests
         changed_paths = []
         request_info = {"exec_root": self._exec_root, "inputs": request["inputs"]}
+        srcdir_prefix = str(srcdir) + "/"
         for entry in request["inputs"]:
             path = entry["path"]
+            # In persistent worker mode, request["inputs"] includes action-level
+            # tools (e.g. sphinx-build, sphinx_build.py) and params files that
+            # live outside srcdir. Only synchronize documentation sources inside srcdir.
+            if not path.startswith(srcdir_prefix):
+                continue
             digest = entry["digest"]
             # Make the path srcdir-relative so Sphinx understands it.
-            path = path.removeprefix(srcdir + "/")
+            path = path.removeprefix(srcdir_prefix)
             incoming_digests[path] = digest
 
             if path not in current_digests:
@@ -145,31 +321,7 @@ def _prepare_sphinx(self, request):
                 logger.info("path %s changed", path)
                 changed_paths.append(path)
 
-        # Remove any source files that were tracked in the previous request (`current_digests`)
-        # but are missing from the current `request["inputs"]` (`incoming_digests`).
-        # Across incremental branch switches or file removals, if these stale symlinks
-        # remain in `srcdir` on disk, Sphinx will discover broken/unreadable files during
-        # `find_files()` and abort with "WARNING: Ignored unreadable document" (fatal with -W).
-        for path in set(current_digests) - set(incoming_digests):
-            removed_path = os.path.join(srcdir, path)
-            if os.path.exists(removed_path) or os.path.islink(removed_path):
-                logger.info("removing stale source file %s", removed_path)
-                try:
-                    if os.path.islink(removed_path):
-                        try:
-                            os.remove(removed_path)
-                        except OSError:
-                            os.rmdir(removed_path)
-                    elif os.path.isdir(removed_path):
-                        shutil.rmtree(removed_path)
-                    else:
-                        os.remove(removed_path)
-                except OSError as e:
-                    logger.warning(
-                        "failed to remove stale source %s: %s", removed_path, e
-                    )
-
-        self._digests[srcdir] = incoming_digests
+        self._digests[str(srcdir)] = incoming_digests
         self._extension.changed_paths = changed_paths
         request_info["changed_sources"] = changed_paths
 
@@ -184,13 +336,22 @@ def _prepare_sphinx(self, request):
         # failures. So on the first request start from a clean slate.
         if is_first_request:
             shutil.rmtree(worker_outdir, ignore_errors=True)
+            shutil.rmtree(destdir, ignore_errors=True)
             for arg in sphinx_args:
                 if arg.startswith("--doctree-dir="):
                     shutil.rmtree(arg.partition("=")[2], ignore_errors=True)
         self._worker_outdirs.add(worker_outdir)
         sphinx_args[1] = worker_outdir
 
-        request_info_path = os.path.join(srcdir, "_bazel_worker_request_info.json")
+        if srcdir not in self._syncers:
+            self._syncers[srcdir] = DirectorySyncer(srcdir, destdir)
+        syncer = self._syncers[srcdir]
+        syncer.sync(incoming_digests)
+
+        sphinx_args[0] = str(destdir)
+        request_info_path = os.path.join(
+            sphinx_args[0], "_bazel_worker_request_info.json"
+        )
         with open(request_info_path, "w") as fp:
             json.dump(request_info, fp)
         sphinx_args.append(f"--define={_REQUEST_INFO_CONFIG_NAME}={request_info_path}")
@@ -341,6 +502,11 @@ def _non_worker_main():
             args.extend(lines)
         else:
             args.append(arg)
+    if len(args) > 1:
+        srcdir = pathlib.Path(args[1])
+        destdir = pathlib.Path(f"{srcdir}.worker-in.d")
+        DirectorySyncer(srcdir, destdir).copytree()
+        args[1] = str(destdir)
     sys.argv[:] = args
     return main()
 

sphinxdocs/tests/sphinx_build/BUILD.bazel

@@ -0,0 +1,11 @@
+load("@rules_python//python:py_test.bzl", "py_test")
+
+py_test(
+    name = "directory_syncer_test",
+    srcs = ["directory_syncer_test.py"],
+    deps = [
+        "//sphinxdocs/private:sphinx_build_lib",
+        "@dev_pip//absl_py",
+        "@dev_pip//sphinx",
+    ],
+)