build/api/ninja_artifacts.py

# Copyright 2024 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Handle the list of Ninja artifacts corresponding to the last build command."""

import collections
import os
import sys
import typing as T
from pathlib import Path

_SCRIPT_DIR = Path(__file__).parent
sys.path.insert(0, str(_SCRIPT_DIR / "../bazel/scripts"))
import build_utils

# The name of the file listing the dependencies of the Ninja build plan,
# as generated by GN. This contains a single statement that looks like:
#
# ```
# build.ninja.stamp: <dep1> <dep2> <dep3> ...
# ```
#
# Where each <dep> points to a BUILD.gn or .gni file, using paths relative
# to the build directory (e.g. `../../src/BUILD.gn`).
#
NINJA_BUILD_PLAN_DEPS_FILE = "build.ninja.d"

# The name of the file that contains the list of targets passed to the
# last `fx build` or `fint build` invocation.
LAST_NINJA_TARGETS_FILE = "last_ninja_build_targets.txt"

# The name of the file that will contain the cached value of all artifacts
# that are transitive inputs for the targets of the last build.
LAST_NINJA_ARTIFACTS_FILE = "last_ninja_artifacts.txt"

# The name of the file that will contain the cached value of all source files
# that are transitive inputs for the targets of the last build.
LAST_NINJA_SOURCES_FILE = "last_ninja_inputs.txt"


class NinjaRunner(object):
    """Wrapper class to invoke Ninja."""

    def __init__(
        self,
        ninja: Path,
        build_dir: Path,
        command_runner: T.Optional[build_utils.CommandRunner] = None,
    ):
        """Create instance.

        Args:
            ninja: Path to Ninja binary.
            build_dir: Path to Ninja build directory.
            command_runner: Optional CommandRunner instance. If None, a default instance will be created.
        """
        self._ninja = ninja
        self._build_dir = build_dir
        self._cmd_runner = (
            command_runner if command_runner else build_utils.CommandRunner()
        )

    @property
    def build_dir(self) -> Path:
        return self._build_dir

    def run_and_extract_output(self, cmd: list[str]) -> str:
        """Run a given Ninja command and return its output.

        Args:
            cmd: list of Ninja options, note that "-C <build_dir>" will be
               prepended to it before invoking Ninja.

        Returns:
           The command's stdout in case of success. stderr is captured but never returned
           unless the command fails (in which case it will be available from the corresponding
           exception object).

        Raises:
           subprocess.CalledProcessError if the command failed.
        """
        ret = self._cmd_runner.run_command(
            [self._ninja, "-C", self._build_dir] + cmd,
            **self._cmd_runner.CAPTURE_KWARGS,
            check=True,
        )
        return ret.stdout


class MockNinjaRunner(NinjaRunner):
    """A mock NinjaRunner instance that can be used in tests."""

    def __init__(self, build_dir: Path, mock_output: str) -> None:
        self._mock_runner = build_utils.MockCommandRunner()
        super().__init__(Path("ninja"), build_dir, self._mock_runner)
        self._mock_runner.push_result(0, mock_output, "")

    def last_ninja_args(self) -> list[str | Path]:
        last_args = self._mock_runner.results[-1].args
        assert last_args[0:3] == ["ninja", "-C", str(self.build_dir)]
        return last_args[3:]


def get_last_build_targets_path(build_dir: Path) -> Path:
    """Return the path of the file listing the targets passed to the last Ninja build."""
    return build_dir / LAST_NINJA_TARGETS_FILE


def get_last_build_targets(build_dir: Path) -> list[str]:
    """Return the list of targets passed to the last Ninja build."""
    last_ninja_targets_path = get_last_build_targets_path(build_dir)
    last_ninja_targets = []
    if last_ninja_targets_path.exists():
        data = last_ninja_targets_path.read_text().strip()
        if data:
            last_ninja_targets = data.split(" ")

    if last_ninja_targets:
        return last_ninja_targets

    # Fallback if the file doesn't exist or is empty.
    return [":default"]


def get_build_plan_deps(build_dir: Path) -> T.Sequence[str]:
    """Return the list of Ninja build plan dependencies.

    This is the list of BUILD.gn or .gni files that, if changed, would trigger
    a regeneration of the Ninja build plan.

    Args:
        build_dir: Ninja build directory path.
    Returns:
        A list of path strings, relative to build_dir.
    """
    deps_path = build_dir / NINJA_BUILD_PLAN_DEPS_FILE
    deps_content = deps_path.read_text().strip()

    # For faster parsing, assert that there are no quoted dependency paths
    # in that file.
    assert (
        deps_content.find("'") < 0
    ), f"Unexpected quoted path in {deps_path.resolve()}"

    deps = deps_content.split(" ")
    assert deps, f"Empty file: {deps_path.resolve()}"

    assert (
        deps[0] == "build.ninja.stamp:"
    ), f"Unexpected {NINJA_BUILD_PLAN_DEPS_FILE} content: {deps[0:4]}"
    return deps[1:]


def check_output_needs_update(
    output_file: Path, input_files: T.Sequence[Path]
) -> bool:
    """Returns True if an output file needs to be updated.

    Args:
      output_file: Output file path.
      input_files: A list of input file paths.
    Returns:
      True if the output_file needs to be re-generated.
    """
    if not output_file.exists():
        return True

    output_timestamp = output_file.stat(follow_symlinks=False).st_mtime
    for input_file in input_files:
        if (
            not input_file.exists()
            or input_file.stat(follow_symlinks=False).st_mtime
            > output_timestamp
        ):
            return True

    return False


def get_last_build_artifacts(ninja_runner: NinjaRunner) -> T.Sequence[str]:
    """Return the list of all Ninja artifacts corresponding to the last `fx` or `fint` build.

    Args:
        ninja_runner: A NinjaRunner instance, used to invoke Ninja if needed.
    Returns:
        A list of Ninja output paths, relative to the build directory.
    """
    build_dir = ninja_runner.build_dir
    last_ninja_targets = get_last_build_targets(build_dir)

    # Determine whether ninja_artifacts.txt needs to be re-generated.
    # This happens when last_ninja_build_targets.txt is modified, or when
    # any of the build plan inputs (e.g. BUILD.gn files) are modified
    # since the last call. These are listed in build.ninja.d which starts
    # with `build.ninja.stamp: ` followed by paths, relative to the
    # build directory.
    ninja_artifacts_path = build_dir / LAST_NINJA_ARTIFACTS_FILE
    ninja_artifacts_deps = [get_last_build_targets_path(build_dir)] + [
        build_dir / dep for dep in get_build_plan_deps(build_dir)
    ]

    if check_output_needs_update(ninja_artifacts_path, ninja_artifacts_deps):
        # Invoke Ninja to regenerate a new set of inputs. Then write results to disk.
        ninja_output = ninja_runner.run_and_extract_output(
            ["-t", "outputs"] + last_ninja_targets
        )
        ninja_artifacts = ninja_output.splitlines()
        ninja_artifacts_path.write_text("\n".join(ninja_artifacts))
    else:
        # Read previous results from disk.
        ninja_artifacts = ninja_artifacts_path.read_text().splitlines()

    return ninja_artifacts


def get_last_build_sources(ninja_runner: NinjaRunner) -> T.Sequence[str]:
    """Return the list of all Ninja sources to the last `fx` or `fint` build.

    Args:
        build_dir: Path to the build directory.
        ninja_runner: A NinjaRunner instance, used to invoke Ninja if needed.
    Returns:
        A list of Ninja input paths, relative to the build directory.
        Note that they will all start with ../.
    """
    build_dir = ninja_runner.build_dir
    last_ninja_targets = get_last_build_targets(build_dir)

    # Determine whether ninja_artifacts.txt needs to be re-generated.
    # This happens when last_ninja_build_targets.txt is modified, or when
    # any of the build plan sources (e.g. BUILD.gn files) are modified
    # since the last call. These are listed in build.ninja.d which starts
    # with `build.ninja.stamp: ` followed by paths, relative to the
    # build directory.
    ninja_sources_path = build_dir / LAST_NINJA_SOURCES_FILE
    ninja_sources_deps = [get_last_build_targets_path(build_dir)] + [
        build_dir / dep for dep in get_build_plan_deps(build_dir)
    ]

    if check_output_needs_update(ninja_sources_path, ninja_sources_deps):
        # Invoke Ninja to regenerate a new set of sources. Then write results to disk.
        ninja_output = ninja_runner.run_and_extract_output(
            ["-t", "inputs", "--no-shell-escape", "--dependency-order"]
            + last_ninja_targets,
        )
        ninja_sources = [
            line for line in ninja_output.splitlines() if line.startswith("../")
        ]
        ninja_sources_path.write_text("\n".join(ninja_sources))
    else:
        # Read previous results from disk.
        ninja_sources = ninja_sources_path.read_text().splitlines()

    return ninja_sources


def should_file_changes_trigger_build(
    changed_files: list[str],
    fuchsia_dir: Path,
    ninja_runner: NinjaRunner,
    root_targets: None | list[str] = None,
) -> tuple[bool, str]:
    """Determine whether file changes should trigger a build.

    Args:
        changed_files: List of file path strings, relative to Fuchsia source directory,
            of files that were changed since the last build.
        fuchsia_dir: Path to Fuchsia source directory.
        ninja_runner: A NinjaRunner instance.
        root_targets: Optional list of Ninja root target paths. If not specified,
            the set of root targets used by the last build will be used.
    Returns:
        A (should_build, reason) pair, where should_build is a boolean flag, and reason is
        a string which will be non-empty when should_build is True, explaining succinctly why
        the build should run.
    """
    changed_sources: set[str] = set()
    for file in changed_files:
        if os.path.isabs(file):
            changed_sources.add(os.path.relpath(file, fuchsia_dir))
        else:
            changed_sources.add(str(file))

    build_dir = ninja_runner.build_dir

    # All source inputs appear with a prefix like ../../ that corresponds
    # to the relative path from the build directory to the Fuchsia source one.
    source_prefix = os.path.relpath(fuchsia_dir, build_dir) + "/"

    # If there are any GN BUILD.gn or .gni files in the input list, check if they are
    # input dependencies of the current Ninja build plan. If this is the case, a new
    # regeneration step must be run to regenerate a new Ninja build graph, making the
    # content of the Ninja deps log potentially incorrect (see https://fxbug,dev/)
    gn_build_files = {s for s in changed_sources if s.endswith((".gn", ".gni"))}
    if gn_build_files:
        # Some plan dependencies are in the Ninja build directory, ignore them.
        current_plan_deps = get_build_plan_deps(build_dir)
        source_plan_deps = set(
            dep[len(source_prefix) :]
            for dep in current_plan_deps
            if dep.startswith(source_prefix)
        )

        changed_plan_deps = gn_build_files & source_plan_deps
        if changed_plan_deps:
            return True, "GN build graph changed."

        changed_sources = changed_sources - gn_build_files

    # Run the Ninja multi-inputs tool to determine the set of source inputs for each
    # one of the last build's targets. This includes all entries from the Ninja deps
    # log, which is more accurate than using GN analyze which doesn't know about them.
    #
    # Note that the GN bazel_action() action generates a depfile that lists all Bazel
    # inputs as well as all Bazel build files used by the corresponding Bazel targets.
    # This will expose these to Ninja through the deps log, which is why there is no
    # need here to invoke Bazel to determine which Bazel targets need to be rebuilt.
    #
    #     Ninja graph                       Bazel graph
    #
    #
    #      :default
    #         |
    #        ...
    #         |
    #         v
    #     some:bazel_action  -----------> @//some/bazel:target, @//some/other/bazel:target
    #          |
    #          |                   bazel_gn_target_action.py invoked from Ninja.
    #          |                   - calls `bazel build` to build artifacts
    #          |                   - calls `bazel query` to get list of inputs and build files
    #          |                   - writes the Bazel inputs + build files paths to depfile,
    #     (.ninja_deps)              using paths relative to Ninja build directory.
    #          |
    #          |
    #          v
    #      ../../some/bazel/BUILD.bazel
    #      ../../some/bazel/source.cc
    #      ../../some/bazel/source.h
    #      ../../some/other/bazel/BUILD.bazel
    #      ../../some/other/bazel/source2.h
    #
    ninja_targets = (
        root_targets
        if root_targets is not None
        else get_last_build_targets(build_dir)
    )

    tool_output = ninja_runner.run_and_extract_output(
        [
            "-t",
            "multi-inputs",
            "--depfile",
        ]
        + ninja_targets
    )

    target_to_sources: dict[str, set[str]] = collections.defaultdict(set)

    for line in tool_output.splitlines():
        # Each line of the tool's output should be
        # <ninja_target> <tab> <ninja_input> <newline>
        target, sep, input = line.partition("\t")
        assert sep == "\t", f"Malformed Ninja tool output line: [{line}]"

        if not input.startswith(source_prefix):
            continue  # Ignore Ninja artifacts

        target_to_sources[target].add(input[len(source_prefix) :])

    updated_targets = []
    for target, sources in target_to_sources.items():
        if bool(sources & changed_sources):
            updated_targets.append(target)

    if not updated_targets:
        return False, ""

    if len(updated_targets) == 1:
        return True, f"Sources updated for target: {updated_targets[0]}"

    return True, f"Sources updated for {len(updated_targets)} targets."