Mark certain types as "side-effecting" (#8922)

## Problem

Pants occasionally needs to perform some side-effecting operation that do not work well with the normal caching semantics provided by @rules. For instance, writing build artifacts to disk, or running a binary locally and interactively.

Over the course of discussing how to go about implementing these kinds of side-effecting operations, we've created a pattern: create a new python type that implements one of these side-effecting operations, and request it as the input to a @goal_rule. Right now, @goal_rules are top-level rules associated with a pants goal that also have the property of being marked as non-cacheable, so it is okay for side-effecting behavior to happen in this subset of rules.

## Solution

This commit formalizes the notion of an side-effecting rule input type by giving the category a name - "side-effecting" - and establishing a check at rule registration time that only non-cacheable @rules are trying to request these types. Right now, the only type of @rule that is non-cacheable is a rule declared as a @goal_rule, but this might change in the future (and would require a corresponding update to the logic of this check).

src/python/pants/backend/python/rules/python_test_runner.py

@@ -15,15 +15,15 @@
 from pants.build_graph.address import Address
 from pants.engine.addressable import BuildFileAddresses
 from pants.engine.fs import Digest, DirectoriesToMerge, FileContent, InputFilesContent
-from pants.engine.interactive_runner import InteractiveProcessRequest, InteractiveRunner
+from pants.engine.interactive_runner import InteractiveProcessRequest
 from pants.engine.isolated_process import ExecuteProcessRequest, FallibleExecuteProcessResult
 from pants.engine.legacy.graph import HydratedTargets, TransitiveHydratedTargets
 from pants.engine.legacy.structs import PythonTestsAdaptor
 from pants.engine.rules import UnionRule, rule, subsystem_rule
 from pants.engine.selectors import Get
 from pants.option.global_options import GlobalOptions
 from pants.rules.core.strip_source_root import SourceRootStrippedSources
-from pants.rules.core.test import TestDebugResult, TestOptions, TestResult, TestTarget
+from pants.rules.core.test import TestDebugRequest, TestOptions, TestResult, TestTarget
 
 
 DEFAULT_COVERAGE_CONFIG = dedent(f"""
@@ -189,17 +189,14 @@ async def run_python_test(
 async def debug_python_test(
   test_target: PythonTestsAdaptor,
   test_setup: TestTargetSetup,
-  runner: InteractiveRunner
-) -> TestDebugResult:
+) -> TestDebugRequest:
 
   run_request = InteractiveProcessRequest(
     argv=(test_setup.requirements_pex.output_filename, *test_setup.args),
     run_in_workspace=False,
     input_files=test_setup.input_files_digest
   )
-
-  result = runner.run_local_interactive_process(run_request)
-  return TestDebugResult(result.process_exit_code)
+  return TestDebugRequest(run_request)
 
 
 def rules():

src/python/pants/engine/console.py

@@ -7,6 +7,7 @@
 from colors import blue, green, red
 
 from pants.engine.native import Native
+from pants.engine.rules import side_effecting
 
 
 #TODO this needs to be a file-like object/stream
@@ -34,8 +35,10 @@ def write(self, payload):
     self._native.write_stderr(self._session, payload)
 
 
+@side_effecting
 class Console:
   """Class responsible for writing text to the console while Pants is running. """
+  side_effecting = True
 
   def __init__(self, stdout=None, stderr=None, use_colors: bool = True, session: Optional[Any] = None):
     """`stdout` and `stderr` may be explicitly provided when Console is constructed. 

src/python/pants/engine/fs.py

@@ -7,7 +7,7 @@
 from typing import TYPE_CHECKING, Iterable, Optional, Tuple
 
 from pants.engine.objects import Collection
-from pants.engine.rules import RootRule
+from pants.engine.rules import RootRule, side_effecting
 from pants.option.custom_types import GlobExpansionConjunction as GlobExpansionConjunction
 from pants.option.global_options import GlobMatchErrorBehavior
 from pants.util.dirutil import (
@@ -209,6 +209,7 @@ class UrlToFetch:
   digest: Digest
 
 
+@side_effecting
 @dataclass(frozen=True)
 class Workspace:
   """Abstract handle for operations that touch the real local filesystem."""

src/python/pants/engine/interactive_runner.py

@@ -6,7 +6,7 @@
 
 from pants.base.exception_sink import ExceptionSink
 from pants.engine.fs import EMPTY_DIRECTORY_DIGEST, Digest
-from pants.engine.rules import RootRule
+from pants.engine.rules import RootRule, side_effecting
 
 
 if TYPE_CHECKING:
@@ -30,6 +30,7 @@ def __post_init__(self):
       raise ValueError("InteractiveProessRequest should use the Workspace API to materialize any needed files when it runs in the workspace")
 
 
+@side_effecting
 @dataclass(frozen=True)
 class InteractiveRunner:
   _scheduler: 'SchedulerSession'

src/python/pants/engine/rules.py

@@ -22,6 +22,13 @@
 from pants.util.meta import frozen_after_init
 
 
+def side_effecting(cls):
+  """Annotates a class to indicate that it is a side-effecting type, which needs
+  to be handled specially with respect to rule caching semantics."""
+  cls.__side_effecting = True
+  return cls
+
+
 class _RuleVisitor(ast.NodeVisitor):
   """Pull `Get` calls out of an @rule body."""
 
@@ -235,9 +242,17 @@ def rule_decorator(*args, **kwargs) -> Callable:
     )
     for name, parameter in signature.parameters.items()
   )
+  validate_parameter_types(func_id, parameter_types, cacheable)
   return _make_rule(return_type, parameter_types, cacheable=cacheable, name=name)(func)
 
 
+def validate_parameter_types(func_id: str, parameter_types: Tuple[Type, ...], cacheable: bool) -> None:
+  if cacheable:
+    for ty in parameter_types:
+      if getattr(ty, "__side_effecting", False):
+        raise ValueError(f"Non-console `@rule` {func_id} has a side-effecting parameter: {ty}")
+
+
 def inner_rule(*args, **kwargs) -> Callable:
   if len(args) == 1 and inspect.isfunction(args[0]):
     return rule_decorator(*args, **kwargs)

src/python/pants/rules/core/test.py

@@ -13,6 +13,7 @@
 from pants.engine.console import Console
 from pants.engine.fs import Digest
 from pants.engine.goal import Goal, GoalSubsystem
+from pants.engine.interactive_runner import InteractiveProcessRequest, InteractiveRunner
 from pants.engine.isolated_process import FallibleExecuteProcessResult
 from pants.engine.legacy.graph import HydratedTarget
 from pants.engine.objects import union
@@ -54,8 +55,8 @@ def from_fallible_execute_process_result(
 
 
 @dataclass(frozen=True)
-class TestDebugResult:
-  exit_code: int
+class TestDebugRequest:
+  ipr: InteractiveProcessRequest
 
 
 @union
@@ -119,17 +120,19 @@ def is_testable(
 
 
 @dataclass(frozen=True)
-class AddressAndDebugResult:
+class AddressAndDebugRequest:
   address: BuildFileAddress
-  test_result: TestDebugResult
+  request: TestDebugRequest
 
 
 @goal_rule
-async def run_tests(console: Console, options: TestOptions, addresses: BuildFileAddresses) -> Test:
+async def run_tests(console: Console, options: TestOptions, runner: InteractiveRunner, addresses: BuildFileAddresses) -> Test:
   if options.values.debug:
     address = await Get[BuildFileAddress](BuildFileAddresses, addresses)
-    result = await Get[AddressAndDebugResult](Address, address.to_address())
-    return Test(result.test_result.exit_code)
+    addr_debug_request = await Get[AddressAndDebugRequest](Address, address.to_address())
+    result = runner.run_local_interactive_process(addr_debug_request.request.ipr)
+    return Test(result.process_exit_code)
+
   results = await MultiGet(Get[AddressAndTestResult](Address, addr.to_address()) for addr in addresses)
   did_any_fail = False
   filtered_results = [(x.address, x.test_result) for x in results if x.test_result is not None]
@@ -186,10 +189,10 @@ async def coordinator_of_tests(
 
 
 @rule
-async def coordinator_of_debug_tests(target: HydratedTarget) -> AddressAndDebugResult:
+async def coordinator_of_debug_tests(target: HydratedTarget) -> AddressAndDebugRequest:
   logger.info(f"Starting tests in debug mode: {target.address.reference()}")
-  result = await Get[TestDebugResult](TestTarget, target.adaptor)
-  return AddressAndDebugResult(target.address, result)
+  request = await Get[TestDebugRequest](TestTarget, target.adaptor)
+  return AddressAndDebugRequest(target.address, request)
 
 
 def rules():

src/python/pants/rules/core/test_test.py

@@ -10,15 +10,16 @@
 from pants.build_graph.address import Address, BuildFileAddress
 from pants.engine.addressable import BuildFileAddresses
 from pants.engine.build_files import AddressProvenanceMap
-from pants.engine.fs import EMPTY_DIRECTORY_DIGEST, Snapshot
+from pants.engine.fs import EMPTY_DIRECTORY_DIGEST, Digest, FileContent, InputFilesContent, Snapshot
+from pants.engine.interactive_runner import InteractiveProcessRequest, InteractiveRunner
 from pants.engine.legacy.graph import HydratedTarget
 from pants.engine.legacy.structs import PythonBinaryAdaptor, PythonTestsAdaptor
 from pants.engine.rules import UnionMembership
 from pants.rules.core.test import (
-  AddressAndDebugResult,
+  AddressAndDebugRequest,
   AddressAndTestResult,
   Status,
-  TestDebugResult,
+  TestDebugRequest,
   TestResult,
   TestTarget,
   coordinator_of_tests,
@@ -35,23 +36,43 @@ def __init__(self, **values):
 
 
 class TestTest(TestBase):
+  def make_ipr(self, content: bytes) -> InteractiveProcessRequest:
+    input_files_content = InputFilesContent((
+      FileContent(path='program.py', content=content, is_executable=True),
+    ))
+    digest = self.request_single_product(Digest, input_files_content)
+    return InteractiveProcessRequest(
+      argv=("/usr/bin/python", "program.py",),
+      run_in_workspace=False,
+      input_files=digest,
+    )
+
+  def make_successful_ipr(self) -> InteractiveProcessRequest:
+    content = b"import sys; sys.exit(0)"
+    return self.make_ipr(content)
+
+  def make_failure_ipr(self) -> InteractiveProcessRequest:
+    content = b"import sys; sys.exit(1)"
+    return self.make_ipr(content)
+
   def single_target_test(self, result, expected_console_output, success=True, debug=False):
     console = MockConsole(use_colors=False)
     options = MockOptions(debug=debug)
+    runner = InteractiveRunner(self.scheduler)
     addr = self.make_build_target_address("some/target")
     res = run_rule(
       run_tests,
-      rule_args=[console, options, BuildFileAddresses([addr])],
+      rule_args=[console, options, runner, BuildFileAddresses([addr])],
       mock_gets=[
         MockGet(
           product_type=AddressAndTestResult,
           subject_type=Address,
           mock=lambda _: AddressAndTestResult(addr, result),
         ),
         MockGet(
-          product_type=AddressAndDebugResult,
+          product_type=AddressAndDebugRequest,
           subject_type=Address,
-          mock=lambda _: AddressAndDebugResult(addr, TestDebugResult(exit_code=0 if success else 1))
+          mock=lambda _: AddressAndDebugRequest(addr, TestDebugRequest(ipr=self.make_successful_ipr() if success else self.make_failure_ipr()))
         ),
         MockGet(
           product_type=BuildFileAddress,
@@ -98,6 +119,7 @@ def test_output_failure(self):
   def test_output_mixed(self):
     console = MockConsole(use_colors=False)
     options = MockOptions(debug=False)
+    runner = InteractiveRunner(self.scheduler)
     target1 = self.make_build_target_address("testprojects/tests/python/pants/passes")
     target2 = self.make_build_target_address("testprojects/tests/python/pants/fails")
 
@@ -110,16 +132,16 @@ def make_result(target):
         raise Exception("Unrecognised target")
       return AddressAndTestResult(target, tr)
 
-    def make_debug_result(target):
-      result = TestDebugResult(exit_code=0 if target == target1 else 1)
-      return AddressAndDebugResult(target, result)
+    def make_debug_request(target):
+      request = TestDebugRequest(ipr=self.make_successful_ipr() if target == target1 else self.make_failure_ipr())
+      return AddressAndDebugRequest(target, request)
 
     res = run_rule(
       run_tests,
-      rule_args=[console, options, (target1, target2)],
+      rule_args=[console, options, runner, (target1, target2)],
       mock_gets=[
         MockGet(product_type=AddressAndTestResult, subject_type=Address, mock=make_result),
-        MockGet(product_type=AddressAndDebugResult, subject_type=Address, mock=make_debug_result),
+        MockGet(product_type=AddressAndDebugRequest, subject_type=Address, mock=make_debug_request),
         MockGet(
           product_type=BuildFileAddress,
           subject_type=AddressSpecs,

tests/python/pants_test/engine/test_rules.py

@@ -86,7 +86,7 @@ async def a_goal_rule_generator(console: Console) -> Example:
   return Example(exit_code=0)
 
 
-class RuleTest(unittest.TestCase):
+class RuleTest(TestBase):
   def test_run_rule_goal_rule_generator(self):
     res = run_rule(
       a_goal_rule_generator,
@@ -95,6 +95,20 @@ def test_run_rule_goal_rule_generator(self):
     )
     self.assertEquals(res, Example(0))
 
+  def test_side_effecting_inputs(self) -> None:
+    @goal_rule
+    def valid_rule(console: Console, b: str) -> Example:
+      return Example(exit_code=0)
+
+    with self.assertRaises(ValueError) as cm:
+      @rule
+      def invalid_rule(console: Console, b: str) -> bool:
+        return False
+
+    error_str = str(cm.exception)
+    assert "invalid_rule has a side-effecting parameter" in error_str
+    assert "pants.engine.console.Console" in error_str
+
 
 class RuleIndexTest(TestBase):
   def test_creation_fails_with_bad_declaration_type(self):