Add symlinked immutable Process inputs (#13848)

As designed by @tdyas in #12716, this change adds `immutable_inputs` to `Process`, and adapts the JVM `@rules` to use them: for now, only for the tools themselves. A followup change will move all JVM classpath inputs to `immutable_inputs`.

Nailgun support is now enabled by specifying the subset of the `immutable_inputs` which should be used to start the server: all others will be used to start the client.

[ci skip-build-wheels]

src/python/pants/backend/java/compile/javac.py

@@ -184,7 +184,8 @@ async def compile_java_source(
                 ),
             ],
             input_digest=merged_digest,
-            use_nailgun=jdk_setup.digest,
+            immutable_input_digests=jdk_setup.immutable_input_digests,
+            use_nailgun=jdk_setup.immutable_input_digests.keys(),
             append_only_caches=jdk_setup.append_only_caches,
             env=jdk_setup.env,
             output_directories=(dest_dir,),

src/python/pants/backend/java/dependency_inference/java_parser.py

@@ -15,7 +15,7 @@
 )
 from pants.backend.java.dependency_inference.types import JavaSourceDependencyAnalysis
 from pants.core.util_rules.source_files import SourceFiles
-from pants.engine.fs import AddPrefix, Digest, DigestContents, MergeDigests
+from pants.engine.fs import AddPrefix, Digest, DigestContents
 from pants.engine.process import BashBinary, FallibleProcessResult, Process, ProcessExecutionFailure
 from pants.engine.rules import Get, MultiGet, collect_rules, rule
 from pants.jvm.jdk_rules import JdkSetup
@@ -84,48 +84,41 @@ async def analyze_java_source_dependencies(
     source_prefix = "__source_to_analyze"
     source_path = os.path.join(source_prefix, source_files.files[0])
     processorcp_relpath = "__processorcp"
+    toolcp_relpath = "__toolcp"
 
-    (
-        tool_classpath,
-        prefixed_processor_classfiles_digest,
-        prefixed_source_files_digest,
-    ) = await MultiGet(
+    (tool_classpath, prefixed_source_files_digest,) = await MultiGet(
         Get(
             MaterializedClasspath,
             MaterializedClasspathRequest(
-                prefix="__toolcp",
                 artifact_requirements=(java_parser_artifact_requirements(),),
             ),
         ),
-        Get(Digest, AddPrefix(processor_classfiles.digest, processorcp_relpath)),
         Get(Digest, AddPrefix(source_files.snapshot.digest, source_prefix)),
     )
 
-    tool_digest = await Get(
-        Digest,
-        MergeDigests(
-            (
-                prefixed_processor_classfiles_digest,
-                tool_classpath.digest,
-                jdk_setup.digest,
-            )
-        ),
-    )
+    immutable_input_digests = {
+        **jdk_setup.immutable_input_digests,
+        toolcp_relpath: tool_classpath.digest,
+        processorcp_relpath: processor_classfiles.digest,
+    }
 
     analysis_output_path = "__source_analysis.json"
 
     process_result = await Get(
         FallibleProcessResult,
         Process(
             argv=[
-                *jdk_setup.args(bash, [*tool_classpath.classpath_entries(), processorcp_relpath]),
+                *jdk_setup.args(
+                    bash, [*tool_classpath.classpath_entries(toolcp_relpath), processorcp_relpath]
+                ),
                 "org.pantsbuild.javaparser.PantsJavaParserLauncher",
                 analysis_output_path,
                 source_path,
             ],
             input_digest=prefixed_source_files_digest,
+            immutable_input_digests=immutable_input_digests,
             output_files=(analysis_output_path,),
-            use_nailgun=tool_digest,
+            use_nailgun=immutable_input_digests.keys(),
             append_only_caches=jdk_setup.append_only_caches,
             env=jdk_setup.env,
             description=f"Analyzing {source_files.files[0]}",

src/python/pants/backend/java/dependency_inference/java_parser_launcher.py

@@ -87,7 +87,6 @@ async def build_processors(bash: BashBinary, jdk_setup: JdkSetup) -> JavaParserC
         MergeDigests(
             (
                 materialized_classpath.digest,
-                jdk_setup.digest,
                 source_digest,
             )
         ),
@@ -108,6 +107,7 @@ async def build_processors(bash: BashBinary, jdk_setup: JdkSetup) -> JavaParserC
             ],
             input_digest=merged_digest,
             append_only_caches=jdk_setup.append_only_caches,
+            immutable_input_digests=jdk_setup.immutable_input_digests,
             env=jdk_setup.env,
             output_directories=(dest_dir,),
             description=f"Compile {_LAUNCHER_BASENAME} import processors with javac",

src/python/pants/backend/java/lint/google_java_format/rules.py

@@ -10,7 +10,7 @@
 from pants.core.goals.fmt import FmtResult
 from pants.core.goals.lint import LintRequest, LintResult, LintResults
 from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
-from pants.engine.fs import Digest, MergeDigests
+from pants.engine.fs import Digest
 from pants.engine.internals.selectors import Get, MultiGet
 from pants.engine.process import BashBinary, FallibleProcessResult, Process, ProcessResult
 from pants.engine.rules import collect_rules, rule
@@ -69,7 +69,6 @@ async def setup_google_java_format(
         Get(
             MaterializedClasspath,
             MaterializedClasspathRequest(
-                prefix="__toolcp",
                 lockfiles=(tool.resolved_lockfile(),),
             ),
         ),
@@ -81,10 +80,11 @@ async def setup_google_java_format(
         else setup_request.request.prior_formatter_result
     )
 
-    input_digest = await Get(
-        Digest,
-        MergeDigests([source_files_snapshot.digest, tool_classpath.digest, jdk_setup.digest]),
-    )
+    toolcp_relpath = "__toolcp"
+    immutable_input_digests = {
+        **jdk_setup.immutable_input_digests,
+        toolcp_relpath: tool_classpath.digest,
+    }
 
     maybe_java16_or_higher_options = []
     if jdk_setup.jre_major_version >= 16:
@@ -97,7 +97,7 @@ async def setup_google_java_format(
         ]
 
     args = [
-        *jdk_setup.args(bash, tool_classpath.classpath_entries()),
+        *jdk_setup.args(bash, tool_classpath.classpath_entries(toolcp_relpath)),
         *maybe_java16_or_higher_options,
         "com.google.googlejavaformat.java.Main",
         *(["--aosp"] if tool.aosp else []),
@@ -107,7 +107,8 @@ async def setup_google_java_format(
 
     process = Process(
         argv=args,
-        input_digest=input_digest,
+        input_digest=source_files_snapshot.digest,
+        immutable_input_digests=immutable_input_digests,
         output_files=source_files_snapshot.files,
         append_only_caches=jdk_setup.append_only_caches,
         env=jdk_setup.env,

src/python/pants/backend/java/package/deploy_jar_test.py

@@ -15,7 +15,6 @@
 from pants.backend.java.target_types import rules as target_types_rules
 from pants.build_graph.address import Address
 from pants.core.goals.package import BuiltPackage
-from pants.engine.fs import Digest, MergeDigests
 from pants.engine.process import BashBinary, Process, ProcessResult
 from pants.jvm import jdk_rules
 from pants.jvm.classpath import rules as classpath_rules
@@ -323,15 +322,15 @@ def _deploy_jar_test(rule_runner: RuleRunner, target_name: str) -> None:
     jdk_setup = rule_runner.request(JdkSetup, [])
     bash = rule_runner.request(BashBinary, [])
 
-    input_digests = rule_runner.request(Digest, [MergeDigests([jdk_setup.digest, fat_jar.digest])])
     process_result = rule_runner.request(
         ProcessResult,
         [
             Process(
                 argv=jdk_setup.args(bash, []) + ("-jar", "dave.jar"),
                 description="Run that test jar",
-                input_digest=input_digests,
+                input_digest=fat_jar.digest,
                 append_only_caches=jdk_setup.append_only_caches,
+                immutable_input_digests=jdk_setup.immutable_input_digests,
                 env=jdk_setup.env,
             )
         ],

src/python/pants/backend/java/test/junit.py

@@ -8,9 +8,9 @@
 from pants.backend.java.target_types import JavaTestSourceField
 from pants.core.goals.test import TestDebugRequest, TestFieldSet, TestResult, TestSubsystem
 from pants.engine.addresses import Addresses
-from pants.engine.fs import Digest, DigestSubset, MergeDigests, PathGlobs, RemovePrefix, Snapshot
+from pants.engine.fs import Digest, DigestSubset, PathGlobs, RemovePrefix, Snapshot
 from pants.engine.process import BashBinary, FallibleProcessResult, Process
-from pants.engine.rules import Get, collect_rules, rule
+from pants.engine.rules import Get, MultiGet, collect_rules, rule
 from pants.engine.unions import UnionRule
 from pants.jvm.classpath import Classpath
 from pants.jvm.jdk_rules import JdkSetup
@@ -40,18 +40,22 @@ async def run_junit_test(
     test_subsystem: TestSubsystem,
     field_set: JavaTestFieldSet,
 ) -> TestResult:
-    classpath = await Get(Classpath, Addresses([field_set.address]))
-    junit_classpath = await Get(
-        MaterializedClasspath,
-        MaterializedClasspathRequest(
-            prefix="__thirdpartycp",
-            lockfiles=(junit.resolved_lockfile(),),
+    classpath, junit_classpath = await MultiGet(
+        Get(Classpath, Addresses([field_set.address])),
+        Get(
+            MaterializedClasspath,
+            MaterializedClasspathRequest(
+                prefix="__thirdpartycp",
+                lockfiles=(junit.resolved_lockfile(),),
+            ),
         ),
     )
-    merged_digest = await Get(
-        Digest,
-        MergeDigests((classpath.content.digest, jdk_setup.digest, junit_classpath.digest)),
-    )
+
+    toolcp_relpath = "__toolcp"
+    immutable_input_digests = {
+        **jdk_setup.immutable_input_digests,
+        toolcp_relpath: junit_classpath.digest,
+    }
 
     reports_dir_prefix = "__reports_dir"
     reports_dir = f"{reports_dir_prefix}/{field_set.address.path_safe_spec}"
@@ -63,7 +67,11 @@ async def run_junit_test(
         Process(
             argv=[
                 *jdk_setup.args(
-                    bash, [*classpath.classpath_entries(), *junit_classpath.classpath_entries()]
+                    bash,
+                    [
+                        *classpath.classpath_entries(),
+                        *junit_classpath.classpath_entries(toolcp_relpath),
+                    ],
                 ),
                 "org.junit.platform.console.ConsoleLauncher",
                 *(("--classpath", user_classpath_arg) if user_classpath_arg else ()),
@@ -72,7 +80,8 @@ async def run_junit_test(
                 reports_dir,
                 *junit.options.args,
             ],
-            input_digest=merged_digest,
+            input_digest=classpath.content.digest,
+            immutable_input_digests=immutable_input_digests,
             output_directories=(reports_dir,),
             append_only_caches=jdk_setup.append_only_caches,
             env=jdk_setup.env,

src/python/pants/backend/scala/compile/scalac.py

@@ -110,11 +110,14 @@ async def compile_scala_source(
             exit_code=0,
         )
 
-    (tool_classpath, merged_transitive_dependency_classpath_entries_digest,) = await MultiGet(
+    toolcp_relpath = "__toolcp"
+    scalac_plugins_relpath = "__plugincp"
+    usercp = "__cp"
+
+    (tool_classpath, merged_transitive_dependency_classpath_entries_digest) = await MultiGet(
         Get(
             MaterializedClasspath,
             MaterializedClasspathRequest(
-                prefix="__toolcp",
                 artifact_requirements=(
                     ArtifactRequirements.from_coordinates(
                         [
@@ -143,32 +146,29 @@ async def compile_scala_source(
         ),
     )
 
-    usercp = "__cp"
     prefixed_transitive_dependency_classpath_digest = await Get(
         Digest, AddPrefix(merged_transitive_dependency_classpath_entries_digest, usercp)
     )
 
-    merged_tool_digest, merged_input_digest = await MultiGet(
-        Get(
-            Digest,
-            MergeDigests(
-                (tool_classpath.digest, scalac_plugins.classpath.digest, jdk_setup.digest)
-            ),
-        ),
-        Get(
-            Digest,
-            MergeDigests(
-                (
-                    prefixed_transitive_dependency_classpath_digest,
-                    *(
-                        sources.snapshot.digest
-                        for _, sources in component_members_and_scala_source_files
-                    ),
-                )
-            ),
+    merged_input_digest = await Get(
+        Digest,
+        MergeDigests(
+            (
+                prefixed_transitive_dependency_classpath_digest,
+                *(
+                    sources.snapshot.digest
+                    for _, sources in component_members_and_scala_source_files
+                ),
+            )
         ),
     )
 
+    immutable_input_digests = {
+        **jdk_setup.immutable_input_digests,
+        toolcp_relpath: tool_classpath.digest,
+        scalac_plugins_relpath: scalac_plugins.classpath.digest,
+    }
+
     classpath_arg = ClasspathEntry.arg(
         ClasspathEntry.closure(direct_dependency_classpath_entries), prefix=usercp
     )
@@ -178,11 +178,11 @@ async def compile_scala_source(
         FallibleProcessResult,
         Process(
             argv=[
-                *jdk_setup.args(bash, tool_classpath.classpath_entries()),
+                *jdk_setup.args(bash, tool_classpath.classpath_entries(toolcp_relpath)),
                 "scala.tools.nsc.Main",
                 "-bootclasspath",
-                ":".join(tool_classpath.classpath_entries()),
-                *scalac_plugins.args(),
+                ":".join(tool_classpath.classpath_entries(toolcp_relpath)),
+                *scalac_plugins.args(scalac_plugins_relpath),
                 *(("-classpath", classpath_arg) if classpath_arg else ()),
                 "-d",
                 output_file,
@@ -194,7 +194,8 @@ async def compile_scala_source(
                 ),
             ],
             input_digest=merged_input_digest,
-            use_nailgun=merged_tool_digest,
+            immutable_input_digests=immutable_input_digests,
+            use_nailgun=immutable_input_digests.keys(),
             output_files=(output_file,),
             description=f"Compile {request.component} with scalac",
             level=LogLevel.DEBUG,

src/python/pants/backend/scala/compile/scalac_plugins.py

@@ -122,8 +122,8 @@ class GlobalScalacPlugins:
     names: tuple[str, ...]
     classpath: MaterializedClasspath
 
-    def args(self) -> Iterator[str]:
-        scalac_plugins_arg = ":".join(self.classpath.classpath_entries())
+    def args(self, prefix: str | None = None) -> Iterator[str]:
+        scalac_plugins_arg = ":".join(self.classpath.classpath_entries(prefix))
         if scalac_plugins_arg:
             yield f"-Xplugin:{scalac_plugins_arg}"
         for name in self.names: