diff --git a/docs/docs/using-pants/remote-caching-and-execution/remote-caching.mdx b/docs/docs/using-pants/remote-caching-and-execution/remote-caching.mdx index f22a6c470c1..510df631e5f 100644 --- a/docs/docs/using-pants/remote-caching-and-execution/remote-caching.mdx +++ b/docs/docs/using-pants/remote-caching-and-execution/remote-caching.mdx @@ -94,6 +94,20 @@ remote_cache_read = true remote_cache_write = true ``` +## Caching of downloads + +In addition to process results, Pants caches URL downloads (`http_source` sources and the external tools Pants itself fetches, such as interpreters, linters and formatters) in the remote cache. The downloaded bytes are stored in the remote store, along with an entry recording that the URL was observed to serve exactly those bytes. A machine with a cold local cache — such as an ephemeral CI runner — is then served a previously-verified download entirely from the remote cache, without contacting the origin server (e.g. GitHub) at all, protecting builds from origin outages. + +This behavior is on by default whenever remote cache reads or writes are enabled, and can be disabled with `[GLOBAL].remote_cache_downloads` (for example, during a cache-corruption investigation, or if your organization audits origin fetches at an egress proxy). A download is only ever served from the cache when some machine previously fetched the same URL and verified the same digest declared in the build; content is always re-verified against that digest as it is fetched. Note that rotating `[GLOBAL].process_execution_cache_namespace` does not affect cached downloads (their content is configuration-independent): use the option or server-side deletion instead. Also note that disabling the option (or deleting entries server-side) stops new remote-cache serving, but does not revoke the local record on a machine that was already served a download: such a machine keeps serving that (URL, digest) pair from its local caches until its local store evicts the file. A strict origin-only posture therefore also requires clearing local caches on warm machines. + +Operational notes: + +- Cache uploads happen in the background at the end of the run. A very short-lived run can exit before an upload completes, in which case the next machine re-downloads from the origin and retries the upload; this self-heals, and is bounded by the number of distinct tools. +- Cached downloads stay warm best on servers which validate that an action result's referenced blobs still exist (e.g. bazel-remote's default completeness checking): there, every cache read refreshes both the entry and the bytes. On fleets where machines with `remote_cache_write` are long-lived (and so rarely re-download), it also helps to run at least one write-enabled builder that cycles its local cache, so evicted entries get re-minted. +- Downloads of `file:` URLs and of URLs containing userinfo (`user:password@`) never participate. Presigned URLs which differ on every run work correctly but create one dead cache entry per distinct URL. +- Downloads using `auth_headers` do participate: if the origin is private, anyone with read access to the remote cache can read the downloaded content. Disable `[GLOBAL].remote_cache_downloads` or `[GLOBAL].remote_cache_write` if this does not suit your cache's trust domain. +- Cache entry lookups use the `[GLOBAL].remote_cache_rpc_timeout_millis` and `[GLOBAL].remote_cache_rpc_concurrency` options, with the retry count coming from `[GLOBAL].remote_store_rpc_retries`; byte transfers use the `[GLOBAL].remote_store_*` options. If the cache is unreachable, each download waits up to (`remote_store_rpc_retries` + 1) × `remote_cache_rpc_timeout_millis` (about 90 seconds with default settings) before falling back to the origin, with a warning throttled by `[GLOBAL].remote_cache_warnings`. If a degraded cache is adding too much latency, disable `[GLOBAL].remote_cache_downloads` (or remote caching) until it recovers. + ## Reference Run `pants help-advanced global` or refer to [Global options](../../../reference/global-options.mdx). Most remote execution and caching options begin with the prefix `--remote`. diff --git a/docs/notes/2.34.x.md b/docs/notes/2.34.x.md index 526438cd48d..e2e164dc591 100644 --- a/docs/notes/2.34.x.md +++ b/docs/notes/2.34.x.md @@ -26,9 +26,12 @@ Published Pants binaries are now compiled with a new `dist` Cargo profile that e Pants option config files are now parsed as TOML 1.1 rather than TOML 1.0. This covers `pants.toml` and any other file named by `[GLOBAL].pants_config_files`, the rcfiles named by `[GLOBAL].pantsrc_files` (`/etc/pantsrc`, `~/.pants.rc` and `.pants.rc` by default), and `.toml` files referenced by `@fromfile` option values. Inline tables may now span multiple lines and end with a trailing comma, strings may use the `\e` and `\xHH` escapes, and times may omit their seconds. TOML 1.1 only adds syntax to TOML 1.0, so existing files continue to parse unchanged. TOML files read by backends, such as `pyproject.toml`, are unaffected. -The `remote_cache_rpc_timeout_millis` & `remote_cache_rpc_concurrency` options are now correctly used again after being silently ignored since 2.19. The documented default for `remote_cache_rpc_timeout_millis` is now `30000` to align with the actual behavior since 2.19. If you used the defaults then nothing changes; if you explicitly set them they now take effect. +The `remote_cache_rpc_timeout_millis` & `remote_cache_rpc_concurrency` options are now correctly used again after being silently ignored since 2.19. The documented default for `remote_cache_rpc_timeout_millis` is now `30000` to align with the actual behavior since 2.19. If you used the defaults then nothing changes; if you explicitly set them they now take effect. Conversely, since 2.19 `remote_store_rpc_timeout_millis` and `remote_store_rpc_concurrency` accidentally governed cache (ActionCache) RPCs as well: if you set the store options to tune cache lookups, set the corresponding `remote_cache_rpc_*` options now, as cache RPCs otherwise revert to the cache options' defaults. + +Fixed a bug where `remote_store_rpc_retries` was silently ignored by the REAPI store and cache providers, which always used 2 retries (since 2.2). If you used the default nothing changes; if you explicitly set it, it now takes effect for both store and cache RPCs — for example, a value of 8 now makes up to 9 attempts (previously 3), and a value of 0 now means a single attempt. Also fixed a related bug where the experimental OpenDAL provider used one more than `remote_store_rpc_retries` as the number of retries. + +Downloads (e.g. `http_source` sources and the external tools Pants itself fetches, such as interpreters and linters) now participate in remote caching. When remote caching is configured, the bytes of a downloaded file are stored in the remote store along with a record that the URL was observed to serve exactly those bytes, so machines with cold local caches (e.g. ephemeral CI runners) are served previously-verified downloads entirely from the remote cache instead of re-fetching from the origin (and failing when, say, GitHub has an outage). Content is always re-verified against the digest declared in the build as it is fetched. This is enabled by default whenever remote cache reads or writes are enabled, and can be disabled with the new `[GLOBAL].remote_cache_downloads` option. -Fixed a bug where `remote_store_rpc_retries` was silently ignored by the REAPI cache provider, which always used 2 retries (since 2.2). If you used the default nothing changes; if you explicitly set it, it now takes effect. Also fixed a related bug where the experimental OpenDAL provider used one more than `remote_store_rpc_retries` as the number of retries. ### Goals diff --git a/src/python/pants/engine/internals/buildbarn_integration_tests/BUILD b/src/python/pants/engine/internals/buildbarn_integration_tests/BUILD index adeacf40378..6f8c66a1e28 100644 --- a/src/python/pants/engine/internals/buildbarn_integration_tests/BUILD +++ b/src/python/pants/engine/internals/buildbarn_integration_tests/BUILD @@ -6,6 +6,7 @@ python_sources(dependencies=[":config"]) python_tests( name="tests", overrides={ + "buildbarn_download_cache_integration_test.py": {"timeout": 300}, "buildbarn_remote_cache_integration_test.py": {"timeout": 180}, "buildbarn_remote_execution_integration_test.py": {"timeout": 300}, }, diff --git a/src/python/pants/engine/internals/buildbarn_integration_tests/buildbarn_download_cache_integration_test.py b/src/python/pants/engine/internals/buildbarn_integration_tests/buildbarn_download_cache_integration_test.py new file mode 100644 index 00000000000..aab0b93e19a --- /dev/null +++ b/src/python/pants/engine/internals/buildbarn_integration_tests/buildbarn_download_cache_integration_test.py @@ -0,0 +1,287 @@ +# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md). +# Licensed under the Apache License, Version 2.0 (see LICENSE). + +from __future__ import annotations + +import hashlib +import shutil +import time +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler + +import pytest + +from pants.engine.fs import ( + CreateDigest, + Digest, + DigestContents, + DownloadFile, + FileContent, + FileDigest, + Snapshot, +) +from pants.engine.internals.buildbarn_integration_tests.stack import ( + CACHE_SPECULATION_DELAY_MILLIS, + CacheOnlyBuildbarn, + LocalBuildbarnStack, + should_skip_for_missing_docker, +) +from pants.engine.internals.scheduler import ExecutionError +from pants.engine.process import Process, ProcessResult +from pants.engine.rules import QueryRule +from pants.testutil.rule_runner import RuleRunner +from pants.util.contextutil import http_server +from pants.util.logging import LogLevel + +pytestmark = pytest.mark.skipif( + should_skip_for_missing_docker(), reason="Docker is required for Buildbarn tests" +) + + +# The file being downloaded, standing in for an external tool release (shellcheck, a +# python-build-standalone interpreter, ...). Downloads always declare the expected digest of the +# file up front (in `known_versions`, `http_source`, etc.), so every path below — origin download +# and cache hit alike — verifies the received bytes against this digest. +FILE_CONTENT = b"#!/bin/sh\necho this stands in for a real external tool release\n" +FILE_DIGEST = FileDigest(hashlib.sha256(FILE_CONTENT).hexdigest(), len(FILE_CONTENT)) + + +@dataclass +class Origin: + """The state of the origin server (standing in for e.g. github.com).""" + + request_count: int = 0 + healthy: bool = True + + +def origin_handler(origin: Origin) -> type[BaseHTTPRequestHandler]: + """An origin HTTP server which counts every request, and which can be taken down (it then + returns 504 for everything, like GitHub during an outage).""" + + class OriginHandler(BaseHTTPRequestHandler): + def do_GET(self): + origin.request_count += 1 + if origin.healthy: + self.send_response(200) + self.send_header("Content-Length", f"{len(FILE_CONTENT)}") + self.end_headers() + self.wfile.write(FILE_CONTENT) + else: + self.send_response(504) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, format, *args): + # Keep request logging out of the test output. + pass + + return OriginHandler + + +def fresh_machine( + buildbarn: CacheOnlyBuildbarn, + *, + remote_cache_downloads: bool = True, + instance_name: str | None = None, +) -> RuleRunner: + """A machine with completely empty local caches (like a fresh, ephemeral CI runner), + configured to read and write the Buildbarn remote cache.""" + return RuleRunner( + rules=[ + QueryRule(Snapshot, [DownloadFile]), + QueryRule(DigestContents, [Digest]), + QueryRule(Digest, [CreateDigest]), + QueryRule(ProcessResult, [Process]), + ], + isolated_local_store=True, + bootstrap_args=[ + "--remote-cache-read", + "--remote-cache-write", + f"--remote-store-address={buildbarn.address}", + f"--remote-instance-name={instance_name or buildbarn.instance_name}", + "--remote-cache-downloads" if remote_cache_downloads else "--no-remote-cache-downloads", + ], + ) + + +def test_downloads_survive_an_origin_outage_via_the_remote_cache() -> None: + """The end-to-end regression test for the issue motivating `[GLOBAL].remote_cache_downloads` + (https://github.com/pantsbuild/pants/issues/16785): CI builds failed whenever GitHub returned + 5xx errors while a machine downloaded an external tool, because downloads never used the + remote cache — a fresh machine had no caching layer between it and the origin, no matter how + many caches were configured. + + The story, against a real Buildbarn remote cache: + + 1. While the origin is healthy, machine A downloads the file, verifies its digest, and + records it in the remote cache. + 2. The origin goes down. + 3. Machine B — brand new, empty local caches — still succeeds, without sending a single + request to the dead origin: the download is served from the remote cache and re-verified + against the expected digest. + 4. Machine C — also brand new, but with download caching disabled, which is exactly how + every prior version of Pants behaved — fails with the origin's 504 despite the remote + cache holding everything it needs. + """ + origin = Origin() + with LocalBuildbarnStack() as buildbarn, http_server(origin_handler(origin)) as port: + download = DownloadFile(f"http://127.0.0.1:{port}/tool-v1.2.3.tar.gz", FILE_DIGEST) + + # 1. Machine A downloads from the healthy origin. The remote cache has never seen this + # URL, so the fetch really does hit the origin... + machine_a = fresh_machine(buildbarn, remote_cache_downloads=True) + snapshot_a = machine_a.request(Snapshot, [download]) + assert snapshot_a.files == ("tool-v1.2.3.tar.gz",) + assert origin.request_count == 1 + assert machine_a.scheduler.get_metrics()["remote_download_cache_requests_uncached"] == 1 + # ...and the verified file is then uploaded to the remote cache in the background. + time.sleep(1) + + # 2. The origin goes down: every request to it now returns 504. + origin.healthy = False + + # 3. Machine B, with empty local caches, downloads the same URL: it succeeds with the + # exact bytes machine A verified, and the dead origin is never contacted. + machine_b = fresh_machine(buildbarn, remote_cache_downloads=True) + snapshot_b = machine_b.request(Snapshot, [download]) + assert snapshot_b == snapshot_a + contents = machine_b.request(DigestContents, [snapshot_b.digest]) + assert [(f.path, f.content) for f in contents] == [("tool-v1.2.3.tar.gz", FILE_CONTENT)] + assert origin.request_count == 1 + assert machine_b.scheduler.get_metrics()["remote_download_cache_requests_cached"] == 1 + + # 4. Machine C behaves like Pants did before download caching existed: it never consults + # the remote cache, so the same download retries against the dead origin and then fails + # the build with the origin's error — the exact failure mode from pants#16785. + machine_c = fresh_machine(buildbarn, remote_cache_downloads=False) + with pytest.raises(ExecutionError) as exc: + machine_c.request(Snapshot, [download]) + assert "Server error (504)" in str(exc.value) + assert origin.request_count > 1 + assert "remote_download_cache_requests" not in machine_c.scheduler.get_metrics() + + +def test_bytes_already_in_the_cas_never_satisfy_a_download() -> None: + """The remote cache only serves a download when some machine previously verified that the + URL itself serves those bytes: file content that is merely present in the CAS (here: + uploaded as the output of a remotely-cached process) is never trusted for a URL. + + This strictness is deliberate. If content-addressed bytes from anywhere could satisfy a + download, a mistyped or dead URL paired with a stale-but-correct digest would silently keep + "working" until the cache evicted the file, and then break some unrelated build much later + (https://github.com/pantsbuild/pants/issues/13255). Instead, each URL is fetched and + digest-verified for real once, and only that verified association is served from the cache. + """ + origin = Origin() + with LocalBuildbarnStack() as buildbarn, http_server(origin_handler(origin)) as port: + download = DownloadFile(f"http://127.0.0.1:{port}/tool-v1.2.3.tar.gz", FILE_DIGEST) + + # A process (not a download) produces a file with the exact bytes the download expects, + # and its output is uploaded to the remote cache. + # NB: Built per machine, so that each machine materializes the process's input into its + # own local store; the process itself (and so its cache key) is identical on every + # machine. + def produce_tool_process(machine: RuleRunner) -> Process: + input_digest = machine.request( + Digest, [CreateDigest([FileContent("input.bin", FILE_CONTENT)])] + ) + return Process( + ["/bin/cp", "input.bin", "tool-v1.2.3.tar.gz"], + description="Produce a file with the same bytes as the download", + input_digest=input_digest, + output_files=["tool-v1.2.3.tar.gz"], + level=LogLevel.INFO, + remote_cache_speculation_delay_millis=CACHE_SPECULATION_DELAY_MILLIS, + ) + + machine_a = fresh_machine(buildbarn) + machine_a.request(ProcessResult, [produce_tool_process(machine_a)]) + time.sleep(1) # Let the background upload to the remote cache land. + + # Premise check: those bytes really are in the remote cache and servable — a fresh + # machine re-running the process is handed the identical file without executing it. + machine_b = fresh_machine(buildbarn) + result_b = machine_b.request(ProcessResult, [produce_tool_process(machine_b)]) + contents = machine_b.request(DigestContents, [result_b.output_digest]) + assert [(f.path, f.content) for f in contents] == [("tool-v1.2.3.tar.gz", FILE_CONTENT)] + assert machine_b.scheduler.get_metrics()["remote_cache_requests_cached"] == 1 + + # And yet downloading a URL that expects that digest still fetches from the origin: + # no machine has ever verified that THIS URL serves those bytes. + machine_c = fresh_machine(buildbarn) + snapshot = machine_c.request(Snapshot, [download]) + assert snapshot.files == ("tool-v1.2.3.tar.gz",) + assert origin.request_count == 1 + assert machine_c.scheduler.get_metrics()["remote_download_cache_requests_uncached"] == 1 + + +def test_evicted_file_content_falls_back_to_the_origin_and_reheals() -> None: + """Remote caches evict: the recorded URL association can outlive the file content it points + to, or the cache can lose data entirely. A machine must then fall back to downloading from + the origin — the build still succeeds — and its fallback re-seeds the cache, so machines + after it are protected again. + """ + origin = Origin() + stack = LocalBuildbarnStack() + with stack as buildbarn, http_server(origin_handler(origin)) as port: + download = DownloadFile(f"http://127.0.0.1:{port}/tool-v1.2.3.tar.gz", FILE_DIGEST) + + # Machine A seeds the remote cache from the healthy origin. + machine_a = fresh_machine(buildbarn) + machine_a.request(Snapshot, [download]) + assert origin.request_count == 1 + time.sleep(1) # Let the background upload to the remote cache land. + + # The cache server loses the downloaded file content: stop it, wipe only its CAS + # storage (keeping the ActionCache storage, where URL associations are recorded), and + # start it again. + stack.stop_cache_service() + cas_storage = buildbarn.temp_dir / "storage-cas" + shutil.rmtree(cas_storage) + (cas_storage / "persistent_state").mkdir(parents=True) + # NB: The restarted service gets a fresh host port, so `buildbarn` must be rebound. + buildbarn = stack.start_cache_service() + + # A cold machine can no longer be served from the cache, so it falls back to the + # origin, and re-uploads the verified file. + machine_b = fresh_machine(buildbarn) + machine_b.request(Snapshot, [download]) + assert origin.request_count == 2 + assert machine_b.scheduler.get_metrics()["remote_download_cache_requests_uncached"] == 1 + time.sleep(1) # Let the re-upload land. + + # That fallback healed the cache: the next cold machine is served from it again, even + # through an origin outage. + origin.healthy = False + machine_c = fresh_machine(buildbarn) + machine_c.request(Snapshot, [download]) + assert origin.request_count == 2 + assert machine_c.scheduler.get_metrics()["remote_download_cache_requests_cached"] == 1 + + +def test_rejected_cache_writes_degrade_gracefully() -> None: + """Some cache servers refuse ActionCache writes from clients — this stack's Buildbarn only + authorizes them for its configured instance name. Downloads must still work: the machine + fetches from the origin and only the background cache write fails (visible in the metrics). + Nothing gets recorded, though, so every fresh machine keeps going to the origin. + """ + origin = Origin() + with LocalBuildbarnStack() as buildbarn, http_server(origin_handler(origin)) as port: + download = DownloadFile(f"http://127.0.0.1:{port}/tool-v1.2.3.tar.gz", FILE_DIGEST) + + # This machine uses an instance name Buildbarn's ActionCache does not allow writes for. + machine_a = fresh_machine(buildbarn, instance_name="unauthorized-instance") + snapshot = machine_a.request(Snapshot, [download]) + assert snapshot.files == ("tool-v1.2.3.tar.gz",) + assert origin.request_count == 1 + time.sleep(1) # Give the background cache write time to fail. + metrics = machine_a.scheduler.get_metrics() + assert metrics["remote_download_cache_write_attempts"] == 1 + assert metrics["remote_download_cache_write_errors"] == 1 + assert "remote_download_cache_write_successes" not in metrics + + # Nothing was recorded, so the next fresh machine downloads from the origin again. + machine_b = fresh_machine(buildbarn, instance_name="unauthorized-instance") + machine_b.request(Snapshot, [download]) + assert origin.request_count == 2 + assert machine_b.scheduler.get_metrics()["remote_download_cache_requests_uncached"] == 1 diff --git a/src/python/pants/engine/internals/scheduler.py b/src/python/pants/engine/internals/scheduler.py index 93bb3052a2e..7e6906a4452 100644 --- a/src/python/pants/engine/internals/scheduler.py +++ b/src/python/pants/engine/internals/scheduler.py @@ -227,6 +227,7 @@ def __init__( local_cache=execution_options.local_cache, remote_cache_read=execution_options.remote_cache_read, remote_cache_write=execution_options.remote_cache_write, + remote_cache_downloads=execution_options.remote_cache_downloads, use_sandboxer=execution_options.use_sandboxer, local_parallelism=execution_options.process_execution_local_parallelism, local_enable_nailgun=execution_options.process_execution_local_enable_nailgun, diff --git a/src/python/pants/option/bootstrap_options.py b/src/python/pants/option/bootstrap_options.py index 6484c27045f..e331c84db45 100644 --- a/src/python/pants/option/bootstrap_options.py +++ b/src/python/pants/option/bootstrap_options.py @@ -621,6 +621,7 @@ class ExecutionOptions: remote_cache_warnings: RemoteCacheWarningsBehavior remote_cache_rpc_concurrency: int remote_cache_rpc_timeout_millis: int + remote_cache_downloads: bool remote_execution_address: str | None remote_execution_headers: dict[str, str] @@ -670,6 +671,7 @@ def from_options( remote_cache_warnings=bootstrap_options.remote_cache_warnings, remote_cache_rpc_concurrency=dynamic_remote_options.cache_rpc_concurrency, remote_cache_rpc_timeout_millis=bootstrap_options.remote_cache_rpc_timeout_millis, + remote_cache_downloads=bootstrap_options.remote_cache_downloads, # Remote execution setup. remote_execution_address=dynamic_remote_options.execution_address, remote_execution_headers=cls.with_user_agent(dynamic_remote_options.execution_headers), @@ -767,6 +769,7 @@ def from_options(cls, options: OptionValueContainer) -> LocalStoreOptions: # NB: Matches the store RPC timeout: from 2.19 until this option's wiring was restored, cache # RPCs accidentally used the store timeout, so this preserves the effective default. remote_cache_rpc_timeout_millis=30000, + remote_cache_downloads=True, # Remote execution setup. remote_execution_address=None, remote_execution_headers={}, @@ -1665,6 +1668,33 @@ class BootstrapOptions: default=DEFAULT_EXECUTION_OPTIONS.remote_cache_rpc_timeout_millis, help="Timeout value for remote cache RPCs in milliseconds.", ) + remote_cache_downloads = BoolOption( + advanced=True, + default=DEFAULT_EXECUTION_OPTIONS.remote_cache_downloads, + help=softwrap( + f""" + Whether to cache URL downloads (for example, `http_source` sources and the tools + Pants itself downloads) in the remote cache. + + When enabled, the bytes of a downloaded file are stored in the remote store, along + with an entry recording that the URL was observed to serve exactly those bytes. A + machine with a cold local cache can then serve a previously-verified download + entirely from the remote cache, without contacting the origin server (e.g. GitHub) + at all. Content is always re-verified against the digest declared in the build as it + is fetched. + + Reading requires `[GLOBAL].remote_cache_read` and writing requires + `[GLOBAL].remote_cache_write`; this option has no effect unless remote caching is + configured. Note that disabling this option stops remote cache use for downloads, + but machines that were already served a download keep using their local caches + for it. + + See {doc_url("docs/using-pants/remote-caching-and-execution/remote-caching")} for + details, including which downloads are excluded, the trust implications for private + origins, and how a degraded cache behaves. + """ + ), + ) remote_execution_address = StrOption( advanced=True, default=cast(str, DEFAULT_EXECUTION_OPTIONS.remote_execution_address), diff --git a/src/python/pants/option/global_options_test.py b/src/python/pants/option/global_options_test.py index eaedcfa3668..4e0571195d5 100644 --- a/src/python/pants/option/global_options_test.py +++ b/src/python/pants/option/global_options_test.py @@ -11,7 +11,7 @@ import pytest from pants.base.build_environment import get_buildroot -from pants.engine.internals.native_engine import PyRemotingOptions +from pants.engine.internals.native_engine import PyExecutionStrategyOptions, PyRemotingOptions from pants.option.bootstrap_options import DynamicRemoteOptions, ExecutionOptions, RemoteProvider from pants.option.errors import OptionsError from pants.option.global_options import GlobalOptions @@ -271,3 +271,19 @@ def test_remote_provider_matches_rust_enum( client_key_path=None, append_only_caches_base_path=None, ) + + +def test_execution_strategy_options_kwargs_match_rust() -> None: + PyExecutionStrategyOptions( + local_cache=True, + remote_cache_read=False, + remote_cache_write=False, + remote_cache_downloads=True, + use_sandboxer=False, + local_parallelism=1, + local_enable_nailgun=False, + remote_parallelism=1, + child_max_memory=0, + child_default_memory=1, + graceful_shutdown_timeout=3, + ) diff --git a/src/rust/Cargo.lock b/src/rust/Cargo.lock index 74e49c61a67..787bbe2deee 100644 --- a/src/rust/Cargo.lock +++ b/src/rust/Cargo.lock @@ -1220,6 +1220,7 @@ dependencies = [ "rand 0.10.0", "regex", "remote", + "remote_provider", "reqwest", "rule_graph", "sandboxer", diff --git a/src/rust/engine/Cargo.toml b/src/rust/engine/Cargo.toml index 3fca26726c6..c3f549a3308 100644 --- a/src/rust/engine/Cargo.toml +++ b/src/rust/engine/Cargo.toml @@ -55,6 +55,7 @@ process_execution = { path = "../process_execution" } pyo3 = { workspace = true } rand = { workspace = true } regex = { workspace = true } +remote_provider = { path = "../remote_provider" } reqwest = { workspace = true, default-features = false, features = ["stream", "rustls"] } rule_graph = { path = "../rule_graph" } sandboxer = { path = "../process_execution/sandboxer" } diff --git a/src/rust/engine/src/context.rs b/src/rust/engine/src/context.rs index 9d0184d202f..35010a1ccca 100644 --- a/src/rust/engine/src/context.rs +++ b/src/rust/engine/src/context.rs @@ -27,6 +27,7 @@ use process_execution::{ use regex::Regex; use remote::remote_cache::{RemoteCacheRunnerOptions, RemoteCacheWarningsBehavior}; use remote::{self, remote_cache}; +use remote_provider::choose_action_cache_provider; use rule_graph::RuleGraph; use sandboxer::Sandboxer; use store::{self, ImmutableInputs, RemoteProvider, RemoteStoreOptions, Store, StoreCliOpt}; @@ -37,6 +38,7 @@ use workunit_store::{Metric, RunningWorkunit}; use crate::nodes::{ExecuteProcess, NodeKey, NodeOutput, NodeResult, SubjectPath}; use crate::python::{Failure, throw}; +use crate::remote_download_cache::RemoteDownloadCache; use crate::session::{Session, Sessions}; use crate::tasks::{Rule, Tasks}; use crate::types::Types; @@ -77,6 +79,11 @@ pub struct Core { /// their outputs, and so should be listed before uncached `CommandRunners`. pub command_runners: Vec>, pub http_client: reqwest::Client, + /// The remote tier of the download caches, present when both remote caching and + /// `remote_cache_downloads` are enabled. Holds the full remote-capable Store even when + /// `Core::store` has been restricted to local-only: the download node is a remote cache code + /// path, exactly like the remote cache CommandRunner. + pub remote_download_cache: Option>, pub local_cache: PersistentCache, pub vfs: FS, pub vfs_system: FS, @@ -168,6 +175,7 @@ pub struct ExecutionStrategyOptions { pub local_enable_nailgun: bool, pub remote_cache_read: bool, pub remote_cache_write: bool, + pub remote_cache_downloads: bool, pub child_max_memory: usize, pub child_default_memory: usize, pub graceful_shutdown_timeout: Duration, @@ -693,6 +701,25 @@ impl Core { None }; + let remote_download_cache = if exec_strategy_opts.remote_cache_downloads + && (exec_strategy_opts.remote_cache_read || exec_strategy_opts.remote_cache_write) + { + let provider = choose_action_cache_provider( + remoting_opts.to_remote_cache_options(tls_config.clone())?, + ) + .await?; + Some(Arc::new(RemoteDownloadCache::new( + provider, + full_store.clone(), + exec_strategy_opts.remote_cache_read, + exec_strategy_opts.remote_cache_write, + remoting_opts.cache_warnings_behavior, + executor.clone(), + ))) + } else { + None + }; + let immutable_inputs = ImmutableInputs::new(store.clone(), &local_execution_root_dir)?; let named_caches = NamedCaches::new_local(named_caches_dir); let command_runners = Self::make_command_runners( @@ -762,6 +789,7 @@ impl Core { sandboxer, command_runners, http_client, + remote_download_cache, local_cache, vfs: FS::new(&build_root, ignorer, executor.clone()) .map_err(|e| format!("Could not initialize Vfs: {e:?}"))?, diff --git a/src/rust/engine/src/downloads.rs b/src/rust/engine/src/downloads.rs index cdd9f52433c..b16e934080e 100644 --- a/src/rust/engine/src/downloads.rs +++ b/src/rust/engine/src/downloads.rs @@ -293,11 +293,38 @@ fn filesize_with_suffix(filesize: usize) -> String { } } +/// Test-only HTTP server plumbing, shared with the `DownloadedFile` node tests. +#[cfg(test)] +pub(crate) mod test_server { + use std::net::SocketAddr; + + use axum::Router; + + /// Serve `router` on an OS-assigned localhost port for the remainder of the test, returning + /// the bound address. Must be called from within a tokio runtime. + pub(crate) fn spawn_test_server(router: Router) -> SocketAddr { + let bind_addr = "127.0.0.1:0".parse::().unwrap(); + let listener = std::net::TcpListener::bind(bind_addr).unwrap(); + // NB: axum_server requires the std listener to be non-blocking. + listener.set_nonblocking(true).unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + axum_server::from_tcp(listener) + .expect("Unable to create Server from std::net::TcpListener") + .serve(router.into_make_service()) + .await + .unwrap(); + }); + + addr + } +} + #[cfg(test)] mod tests { use std::{ collections::{BTreeMap, HashSet}, - net::SocketAddr, num::NonZeroUsize, sync::{ Arc, @@ -315,6 +342,7 @@ mod tests { use url::Url; use workunit_store::WorkunitStore; + use super::test_server::spawn_test_server; use super::{download, filesize_with_suffix}; const TEST_RESPONSE: &[u8] = b"xyzzy"; @@ -326,20 +354,8 @@ mod tests { let dir = TempDir::new().unwrap(); let store = Store::local_only(task_executor::Executor::new(), dir.path()).unwrap(); - let bind_addr = "127.0.0.1:0".parse::().unwrap(); - let listener = std::net::TcpListener::bind(bind_addr).unwrap(); - listener.set_nonblocking(true).unwrap(); - let addr = listener.local_addr().unwrap(); - let router = Router::new().route("/foo.txt", get(|| async { TEST_RESPONSE })); - - tokio::spawn(async move { - axum_server::from_tcp(listener) - .expect("Unable to create Server from std::net::TcpListener") - .serve(router.into_make_service()) - .await - .unwrap(); - }); + let addr = spawn_test_server(router); let http_client = reqwest::Client::new(); let url = Url::parse(&format!("http://127.0.0.1:{}/foo.txt", addr.port())).unwrap(); @@ -372,11 +388,6 @@ mod tests { let dir = TempDir::new().unwrap(); let store = Store::local_only(task_executor::Executor::new(), dir.path()).unwrap(); - let bind_addr = "127.0.0.1:0".parse::().unwrap(); - let listener = std::net::TcpListener::bind(bind_addr).unwrap(); - listener.set_nonblocking(true).unwrap(); - let addr = listener.local_addr().unwrap(); - #[derive(Clone)] struct HandlerState { attempt: Arc, @@ -403,14 +414,7 @@ mod tests { .with_state(HandlerState { attempt: Arc::clone(&attempt), }); - - tokio::spawn(async move { - axum_server::from_tcp(listener) - .expect("Unable to create Server from std::net::TcpListener") - .serve(router.into_make_service()) - .await - .unwrap(); - }); + let addr = spawn_test_server(router); let http_client = reqwest::Client::new(); let url = Url::parse(&format!("http://127.0.0.1:{}/foo.txt", addr.port())).unwrap(); diff --git a/src/rust/engine/src/externs/interface.rs b/src/rust/engine/src/externs/interface.rs index de0aaed4083..b97c80c63d1 100644 --- a/src/rust/engine/src/externs/interface.rs +++ b/src/rust/engine/src/externs/interface.rs @@ -300,6 +300,7 @@ impl PyExecutionStrategyOptions { local_enable_nailgun: bool, remote_cache_read: bool, remote_cache_write: bool, + remote_cache_downloads: bool, child_default_memory: usize, child_max_memory: usize, graceful_shutdown_timeout: usize, @@ -312,6 +313,7 @@ impl PyExecutionStrategyOptions { local_enable_nailgun, remote_cache_read, remote_cache_write, + remote_cache_downloads, child_default_memory, child_max_memory, graceful_shutdown_timeout: Duration::from_secs( diff --git a/src/rust/engine/src/lib.rs b/src/rust/engine/src/lib.rs index 7e0902bfded..fdb84fa602b 100644 --- a/src/rust/engine/src/lib.rs +++ b/src/rust/engine/src/lib.rs @@ -15,6 +15,7 @@ mod interning; mod intrinsics; mod nodes; mod python; +mod remote_download_cache; mod scheduler; mod session; mod tasks; diff --git a/src/rust/engine/src/nodes/downloaded_file.rs b/src/rust/engine/src/nodes/downloaded_file.rs index 11e3f0922ea..bf648115c70 100644 --- a/src/rust/engine/src/nodes/downloaded_file.rs +++ b/src/rust/engine/src/nodes/downloaded_file.rs @@ -7,79 +7,101 @@ use std::sync::Arc; use std::time::Duration; use bytes::Bytes; +use cache::PersistentCache; use deepsize::DeepSizeOf; use fs::RelativePath; use graph::CompoundNode; -use grpc_util::prost::MessageExt; -use hashing::Digest; -use protos::pb::pants::cache::{CacheKey, CacheKeyType, ObservedUrl}; use pyo3::prelude::Python; +use store::Store; +use task_executor::TailTasks; use url::Url; use super::{NodeKey, NodeResult}; -use crate::context::{Context, Core}; +use crate::context::Context; use crate::downloads; use crate::externs; use crate::externs::fs::PyFileDigest; use crate::python::{Key, throw}; +use crate::remote_download_cache::{ + RemoteDownloadCache, observed_url_key, url_is_cacheable_remotely, +}; #[derive(Clone, Debug, DeepSizeOf, Eq, Hash, PartialEq)] pub struct DownloadedFile(pub Key); -impl DownloadedFile { - fn url_key(url: &Url, digest: Digest) -> CacheKey { - let observed_url = ObservedUrl { - url: url.as_str().to_owned(), - observed_digest: Some(digest.into()), - }; - CacheKey { - key_type: CacheKeyType::Url.into(), - digest: Some(Digest::of_bytes(&observed_url.to_bytes()).into()), - } - } +/// The state of the machine a download runs against: its local stores and caches, HTTP client, +/// and (when remote caching is configured) the remote download cache. Grouped separately from +/// the per-download arguments so that tests can assemble one without building a full `Core`. +pub(crate) struct DownloadDeps<'a> { + pub(crate) local_cache: &'a PersistentCache, + pub(crate) store: Store, + pub(crate) http_client: &'a reqwest::Client, + pub(crate) remote_download_cache: Option<&'a Arc>, + pub(crate) tail_tasks: TailTasks, + pub(crate) build_id: &'a str, +} - pub async fn load_or_download( - &self, - core: Arc, - url: Url, - auth_headers: BTreeMap, - digest: hashing::Digest, - retry_delay_duration: Duration, - max_attempts: NonZeroUsize, - ) -> Result { - let file_name = url - .path_segments() - .and_then(Iterator::last) - .map(str::to_owned) - .ok_or_else(|| format!("Error getting the file name from the parsed URL: {url}"))?; - let path = RelativePath::new(&file_name).map_err(|e| { - format!( - "The file name derived from {} was {} which is not relative: {:?}", - url, file_name, e - ) - })?; - - // See if we have observed this URL and Digest before: if so, see whether we already have the - // Digest fetched. The extra layer of indirection through the PersistentCache is to sanity - // check that a Digest has ever been observed at the given URL. - // NB: The auth_headers are not part of the key. - let url_key = Self::url_key(&url, digest); - let have_observed_url = core.local_cache.load(&url_key).await?.is_some(); - - // If we hit the ObservedUrls cache, then we have successfully fetched this Digest from - // this URL before. If we still have the bytes, then we skip fetching the content again. - let usable_in_store = have_observed_url - && (core - .store() - .load_file_bytes_with(digest, |_| ()) - .await - .is_ok()); +pub(crate) async fn load_or_download( + deps: DownloadDeps<'_>, + url: Url, + auth_headers: BTreeMap, + digest: hashing::Digest, + retry_delay_duration: Duration, + max_attempts: NonZeroUsize, +) -> Result { + let DownloadDeps { + local_cache, + store, + http_client, + remote_download_cache, + tail_tasks, + build_id, + } = deps; + let file_name = url + .path_segments() + .and_then(Iterator::last) + .map(str::to_owned) + .ok_or_else(|| format!("Error getting the file name from the parsed URL: {url}"))?; + let path = RelativePath::new(&file_name).map_err(|e| { + format!( + "The file name derived from {} was {} which is not relative: {:?}", + url, file_name, e + ) + })?; + + // See if we have observed this URL and Digest before: if so, see whether we already have the + // Digest fetched. The extra layer of indirection through the PersistentCache is to sanity + // check that a Digest has ever been observed at the given URL. + // NB: The auth_headers are not part of the key. + let url_key = observed_url_key(&url, digest); + let have_observed_url = local_cache.load(&url_key).await?.is_some(); + + // If we hit the ObservedUrls cache, then we have successfully fetched this Digest from + // this URL before. If we still have the bytes, then we skip fetching the content again. + // NB: When the node's store handle is remote-capable (remote execution, or + // `cache_content_behavior != fetch`), this probe itself backfills locally-evicted bytes from + // the remote CAS — pre-existing behavior, which bypasses the remote download cache's + // metrics, warning throttling, and marker re-assert below. + let usable_in_store = + have_observed_url && (store.load_file_bytes_with(digest, |_| ()).await.is_ok()); + + if !usable_in_store { + let remote_download_cache = + remote_download_cache.filter(|_| url_is_cacheable_remotely(&url)); + + // The local caches cannot serve this download: consult the remote cache (when + // configured), which serves only (URL, digest) pairs some machine genuinely fetched and + // digest-verified. + let served_remotely = match remote_download_cache { + Some(cache) => cache.load_cached_download(&url, digest, build_id).await, + None => false, + }; - if !usable_in_store { + if !served_remotely { downloads::download( - &core.http_client, - core.store(), - url, + http_client, + store.clone(), + url.clone(), auth_headers, file_name, digest, @@ -87,13 +109,24 @@ impl DownloadedFile { max_attempts, ) .await?; - // The value was successfully fetched and matched the digest: record in the ObservedUrls - // cache. - core.local_cache.store(&url_key, Bytes::from("")).await?; } - core.store().snapshot_of_one_file(path, digest, true).await + // The value was successfully fetched and matched the digest (from the origin, or from the + // remote cache, where it was recorded by a machine which fetched it from the origin): + // record in the ObservedUrls cache. + local_cache.store(&url_key, Bytes::from("")).await?; + if let Some(remote_download_cache) = remote_download_cache { + // Record the verified association remotely in the background. When the download was + // served from the remote cache this re-asserts the existing marker (deliberately + // reusing the full write-back for simplicity), which refreshes it on servers whose + // action-cache entries can be overwritten (e.g. REAPI); on stores with immutable + // entries (e.g. the GitHub Actions cache) the re-assert is a no-op. + remote_download_cache.spawn_write_back(tail_tasks, url, digest); + } } + store.snapshot_of_one_file(path, digest, true).await +} +impl DownloadedFile { pub(super) async fn run_node(self, context: Context) -> NodeResult { let (url_str, expected_digest, auth_headers, retry_delay_duration, max_attempts) = Python::attach(|py| { @@ -120,8 +153,15 @@ impl DownloadedFile { let url = Url::parse(&url_str) .map_err(|err| throw(format!("Error parsing URL {url_str}: {err}")))?; - self.load_or_download( - context.core.clone(), + load_or_download( + DownloadDeps { + local_cache: &context.core.local_cache, + store: context.core.store(), + http_client: &context.core.http_client, + remote_download_cache: context.core.remote_download_cache.as_ref(), + tail_tasks: context.session.tail_tasks(), + build_id: context.session.build_id(), + }, url, auth_headers, expected_digest, @@ -142,3 +182,436 @@ impl From for NodeKey { NodeKey::DownloadedFile(n) } } + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + use std::num::NonZeroUsize; + use std::sync::{ + Arc, + atomic::{AtomicU32, Ordering}, + }; + use std::time::{Duration, Instant}; + + use axum::http::{HeaderMap, header::AUTHORIZATION}; + use axum::{Router, extract::State, response::IntoResponse, routing::get}; + use bytes::Bytes; + use cache::PersistentCache; + use hashing::Digest; + use reqwest::StatusCode; + use store::{RemoteProvider, RemoteStoreOptions, Store}; + use task_executor::TailTasks; + use tempfile::TempDir; + use testutil_mock::{RequestType, StubCAS}; + use url::Url; + use workunit_store::WorkunitStore; + + use super::{DownloadDeps, load_or_download}; + use crate::downloads::test_server::spawn_test_server; + use crate::remote_download_cache::test_util::{make_download_cache, remote_store_options}; + use crate::remote_download_cache::{RemoteDownloadCache, observed_url_key}; + + const TEST_RESPONSE: &[u8] = b"the downloaded file bytes"; + + /// An origin HTTP server which succeeds for the first `successes` requests and returns 504 + /// for every request after that, counting all requests it receives. + fn start_origin(successes: u32) -> (Url, Arc) { + #[derive(Clone)] + struct HandlerState { + requests: Arc, + successes: u32, + } + + let requests = Arc::new(AtomicU32::new(0)); + let router = Router::new() + .route( + "/file.txt", + get(move |State(state): State| async move { + let request = state.requests.fetch_add(1, Ordering::SeqCst); + if request < state.successes { + (StatusCode::OK, TEST_RESPONSE).into_response() + } else { + (StatusCode::GATEWAY_TIMEOUT, &b"504"[..]).into_response() + } + }), + ) + .with_state(HandlerState { + requests: requests.clone(), + successes, + }); + + let addr = spawn_test_server(router); + let url = Url::parse(&format!("http://127.0.0.1:{}/file.txt", addr.port())).unwrap(); + (url, requests) + } + + /// An origin like `start_origin`, but private: requests must carry exactly this + /// `Authorization` header, or they are rejected with a 401. All requests are counted. + fn start_authed_origin(required_authorization: &'static str) -> (Url, Arc) { + #[derive(Clone)] + struct HandlerState { + requests: Arc, + required_authorization: &'static str, + } + + let requests = Arc::new(AtomicU32::new(0)); + let router = Router::new() + .route( + "/file.txt", + get( + move |State(state): State, headers: HeaderMap| async move { + state.requests.fetch_add(1, Ordering::SeqCst); + if headers + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + == Some(state.required_authorization) + { + (StatusCode::OK, TEST_RESPONSE).into_response() + } else { + (StatusCode::UNAUTHORIZED, &b"401"[..]).into_response() + } + }, + ), + ) + .with_state(HandlerState { + requests: requests.clone(), + required_authorization, + }); + + let addr = spawn_test_server(router); + let url = Url::parse(&format!("http://127.0.0.1:{}/file.txt", addr.port())).unwrap(); + (url, requests) + } + + /// The state of a single machine: a local store and ObservedUrls cache (both cold), plus, + /// when a StubCAS is given, a remote download cache configured the way `Core::new` does it in + /// the default remote caching configuration: the machine's own store is the local-only view, + /// while the remote download cache holds the full remote-capable store. + struct Pod { + _store_dir: TempDir, + _cache_dir: TempDir, + store: Store, + local_cache: PersistentCache, + remote_download_cache: Option>, + http_client: reqwest::Client, + tail_tasks: TailTasks, + } + + impl Pod { + async fn new(cas: Option<&StubCAS>, cache_read: bool, cache_write: bool) -> Pod { + match cas { + Some(cas) => { + Self::new_with_remote_options( + remote_store_options(RemoteProvider::Reapi, cas.address()), + cache_read, + cache_write, + ) + .await + } + None => { + let executor = task_executor::Executor::new(); + let (cache_dir, local_cache) = Self::make_local_cache(&executor); + let store_dir = TempDir::new().unwrap(); + let store = Store::local_only(executor, store_dir.path()).unwrap(); + Pod { + _store_dir: store_dir, + _cache_dir: cache_dir, + store, + local_cache, + remote_download_cache: None, + http_client: reqwest::Client::new(), + tail_tasks: TailTasks::new(), + } + } + } + } + + /// As `new(Some(cas), ..)`, but with custom remote store options (e.g. a short RPC + /// timeout). + async fn new_with_remote_options( + options: RemoteStoreOptions, + cache_read: bool, + cache_write: bool, + ) -> Pod { + let executor = task_executor::Executor::new(); + let (cache_dir, local_cache) = Self::make_local_cache(&executor); + let (store_dir, full_store, remote_download_cache) = + make_download_cache(options, cache_read, cache_write).await; + Pod { + _store_dir: store_dir, + _cache_dir: cache_dir, + store: full_store.into_local_only(), + local_cache, + remote_download_cache: Some(remote_download_cache), + http_client: reqwest::Client::new(), + tail_tasks: TailTasks::new(), + } + } + + fn make_local_cache(executor: &task_executor::Executor) -> (TempDir, PersistentCache) { + let cache_dir = TempDir::new().unwrap(); + let local_cache = PersistentCache::new( + cache_dir.path(), + 50 * 1024 * 1024, + executor.clone(), + Duration::from_secs(2 * 60 * 60), + 1, + ) + .unwrap(); + (cache_dir, local_cache) + } + + fn deps(&self) -> DownloadDeps<'_> { + DownloadDeps { + local_cache: &self.local_cache, + store: self.store.clone(), + http_client: &self.http_client, + remote_download_cache: self.remote_download_cache.as_ref(), + tail_tasks: self.tail_tasks.clone(), + build_id: "build_id", + } + } + + async fn download(&self, url: &Url, digest: Digest) -> Result { + self.download_with_headers(url, digest, BTreeMap::new()) + .await + } + + async fn download_with_headers( + &self, + url: &Url, + digest: Digest, + auth_headers: BTreeMap, + ) -> Result { + load_or_download( + self.deps(), + url.clone(), + auth_headers, + digest, + Duration::from_millis(10), + NonZeroUsize::new(1).unwrap(), + ) + .await + } + + // NB: `TailTasks::wait` consumes the shared inner task set, so this is single-shot: a + // second call on the same Pod returns immediately without waiting for anything. + async fn wait_for_write_back(&self) { + self.tail_tasks.clone().wait(Duration::from_secs(10)).await; + } + + async fn assert_bytes_are_local(&self, digest: Digest) { + let loaded = self + .store + .clone() + .into_local_only() + .load_file_bytes_with(digest, Bytes::copy_from_slice) + .await + .unwrap(); + assert_eq!(loaded, Bytes::from_static(TEST_RESPONSE)); + } + } + + fn test_digest() -> Digest { + Digest::of_bytes(TEST_RESPONSE) + } + + #[tokio::test] + async fn cold_pod_is_served_from_remote_cache_when_origin_is_down() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + // The origin serves exactly one request, then returns 504s (a GitHub incident). + let (url, origin_requests) = start_origin(1); + let digest = test_digest(); + + // Pod A downloads from the live origin and writes back. + let pod_a = Pod::new(Some(&cas), true, true).await; + pod_a.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + pod_a.wait_for_write_back().await; + assert!(cas.contains(digest.hash)); + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 1); + + // A cold pod B succeeds via the remote cache: the origin is never contacted. + let pod_b = Pod::new(Some(&cas), true, true).await; + let snapshot = pod_b.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + assert_eq!(snapshot.files(), vec![std::path::PathBuf::from("file.txt")]); + pod_b.assert_bytes_are_local(digest).await; + + // Serving from the remote cache re-asserted the marker (the flow-2 re-assert). + pod_b.wait_for_write_back().await; + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 2); + + // A second download on pod B is served by its local caches: no new AC or origin requests. + let ac_gets = cas.request_count(RequestType::ACGetActionResult); + pod_b.download(&url, digest).await.unwrap(); + assert_eq!(cas.request_count(RequestType::ACGetActionResult), ac_gets); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn cas_blob_without_marker_still_downloads_from_origin() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let digest = test_digest(); + // The digest is already in the CAS (e.g. as a process output), but no marker exists: + // strict semantics require the real download, which then mints the marker. + let cas = StubCAS::builder() + .unverified_content(digest.hash, Bytes::from_static(TEST_RESPONSE)) + .build() + .await; + let (url, origin_requests) = start_origin(1); + + let pod = Pod::new(Some(&cas), true, true).await; + pod.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + // The AC was consulted and correctly missed: the origin download above happened despite + // the blob being present in the CAS, not because the read path was skipped. + assert_eq!(cas.request_count(RequestType::ACGetActionResult), 1); + + pod.wait_for_write_back().await; + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 1); + } + + #[tokio::test] + async fn evicted_blob_falls_back_to_origin_and_reuploads() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let (url, origin_requests) = start_origin(2); + let digest = test_digest(); + + let pod_a = Pod::new(Some(&cas), true, true).await; + pod_a.download(&url, digest).await.unwrap(); + pod_a.wait_for_write_back().await; + + // The blob is evicted from the remote store, but the marker survives. + assert!(cas.remove(digest.hash)); + + // A cold pod falls back to the origin, then restores the marker/blob pairing: the blob + // is re-uploaded AND the marker is re-written, not merely one of the two. + let pod_b = Pod::new(Some(&cas), true, true).await; + pod_b.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 2); + pod_b.wait_for_write_back().await; + assert!(cas.contains(digest.hash)); + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 2); + } + + #[tokio::test] + async fn hung_cache_delays_but_does_not_break_downloads() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + // A degraded cache server which hangs on every lookup (e.g. a blackholed host), for far + // longer than the client's configured RPC timeout. The origin is healthy. + let cas = StubCAS::builder() + .ac_read_delay(Duration::from_secs(60)) + .build() + .await; + let (url, origin_requests) = start_origin(1); + let digest = test_digest(); + + let mut options = remote_store_options(RemoteProvider::Reapi, cas.address()); + options.timeout = Duration::from_millis(250); + let pod = Pod::new_with_remote_options(options, true, true).await; + + // The download waits out the cache lookup's timeout budget, then falls back to the + // origin and succeeds: a degraded cache costs latency, never the build. + let start = Instant::now(); + pod.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + assert!(start.elapsed() < Duration::from_secs(10)); + } + + #[tokio::test] + async fn auth_headers_reach_the_origin_and_are_excluded_from_the_cache_key() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + // A private origin, e.g. one addressed via the S3 URLDownloadHandler (which signs + // requests via auth_headers): it rejects requests without the right credentials. + let (url, origin_requests) = start_authed_origin("Bearer org-token"); + let digest = test_digest(); + + // Pod A's download succeeds only because its auth header genuinely reached the origin + // (the origin 401s otherwise), and the verified file is then written back to the shared + // remote cache: auth-header downloads participate in remote caching. + let pod_a = Pod::new(Some(&cas), true, true).await; + let auth_headers = + BTreeMap::from([("Authorization".to_owned(), "Bearer org-token".to_owned())]); + pod_a + .download_with_headers(&url, digest, auth_headers) + .await + .unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + pod_a.wait_for_write_back().await; + assert!(cas.contains(digest.hash)); + + // Auth headers are deliberately not part of the download cache key (local or remote): a + // cold pod with no credentials at all is served the private origin's bytes from the + // cache, without the origin ever being contacted. (This is the documented trust + // implication of including auth-header downloads: the shared cache is the trust domain.) + let pod_b = Pod::new(Some(&cas), true, true).await; + pod_b.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + pod_b.assert_bytes_are_local(digest).await; + } + + #[tokio::test] + async fn file_urls_never_touch_the_remote_cache() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + + let tempdir = TempDir::new().unwrap(); + let file_path = tempdir.path().join("file.txt"); + std::fs::write(&file_path, TEST_RESPONSE).unwrap(); + let url = Url::parse(&format!("file:{}", file_path.display())).unwrap(); + let digest = test_digest(); + + let pod = Pod::new(Some(&cas), true, true).await; + pod.download(&url, digest).await.unwrap(); + pod.wait_for_write_back().await; + + assert_eq!(cas.request_count(RequestType::ACGetActionResult), 0); + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 0); + assert!(!cas.contains(digest.hash)); + } + + #[tokio::test] + async fn local_marker_hit_makes_no_remote_requests() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + // The origin is hard down. + let (url, origin_requests) = start_origin(0); + let digest = test_digest(); + + // The pod has both the bytes and the ObservedUrls marker: the warm path. + let pod = Pod::new(Some(&cas), true, true).await; + pod.store + .store_file_bytes(Bytes::from_static(TEST_RESPONSE), true) + .await + .unwrap(); + pod.local_cache + .store(&observed_url_key(&url, digest), Bytes::from("")) + .await + .unwrap(); + + pod.download(&url, digest).await.unwrap(); + pod.wait_for_write_back().await; + assert_eq!(origin_requests.load(Ordering::SeqCst), 0); + assert_eq!(cas.request_count(RequestType::ACGetActionResult), 0); + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 0); + } + + #[tokio::test] + async fn remote_caching_disabled_downloads_from_origin() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let (url, origin_requests) = start_origin(1); + let digest = test_digest(); + + let pod = Pod::new(None, false, false).await; + pod.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + + // The second download is served by the local caches. + pod.download(&url, digest).await.unwrap(); + assert_eq!(origin_requests.load(Ordering::SeqCst), 1); + } +} diff --git a/src/rust/engine/src/remote_download_cache.rs b/src/rust/engine/src/remote_download_cache.rs new file mode 100644 index 00000000000..2ddbac467b2 --- /dev/null +++ b/src/rust/engine/src/remote_download_cache.rs @@ -0,0 +1,769 @@ +// Copyright 2026 Pants project contributors (see CONTRIBUTORS.md). +// Licensed under the Apache License, Version 2.0 (see LICENSE). + +//! A remote tier for the two local download caches: downloaded file bytes are stored in the +//! remote CAS, and the verified URL→digest association (the remote analog of the local +//! ObservedUrls cache) is recorded as a synthetic, never-executed REAPI action in the remote +//! ActionCache (the "AC marker"). +//! +//! The marker is the sole remote authority for skipping the URL fetch: if the marker misses but +//! the digest happens to already be in the CAS, we still perform the real download. This keeps +//! the ObservedUrls verification invariant airtight fleet-wide: verification is only ever skipped +//! for (URL, digest) pairs some machine genuinely fetched and digest-verified. + +use std::collections::HashSet; +use std::sync::Arc; + +use futures::FutureExt; +use grpc_util::prost::MessageExt; +use hashing::Digest; +use protos::pb::build::bazel::remote::execution::v2 as remexec; +use protos::pb::pants::cache::{CacheKey, CacheKeyType, ObservedUrl}; +use protos::require_digest; +use remote::remote_cache::{CacheErrorThrottle, CacheErrorType, RemoteCacheWarningsBehavior}; +use remote_provider::ActionCacheProvider; +use store::{Store, StoreError}; +use task_executor::{Executor, TailTasks}; +use url::Url; +use workunit_store::{Level, Metric, in_workunit}; + +/// Discriminates the synthetic download actions from real process executions, and doubles as a +/// format-version salt: bump the version to evolve the encoding without serving stale markers. +const MARKER_ARGUMENTS: [&str; 2] = ["__pants_url_download__", "v1"]; + +/// The path the downloaded file is exposed under in the synthetic ActionResult. Readers use the +/// marker only as an existence proof, so the path is never materialized. +const MARKER_OUTPUT_PATH: &str = "file"; + +/// Downloads carrying URL userinfo (`https://user:token@host/...`) may embed secrets, and +/// `file:` URLs name machine-local paths that other machines cannot fetch or verify: both are +/// excluded from remote read and write entirely. `RemoteDownloadCache`'s methods enforce this +/// themselves; callers may additionally pre-filter with this function as an optimization. +/// +/// NB: Query strings are included even though presigned URLs churn markers: they are part of the +/// local ObservedUrls cache key, and the local and remote keys must agree. +pub fn url_is_cacheable_remotely(url: &Url) -> bool { + url.scheme() != "file" && url.username().is_empty() && url.password().is_none() +} + +/// The local ObservedUrls cache key for a verified (URL, digest) pair: the local tier of the +/// same association `make_marker_command` (directly below) encodes remotely. Both are +/// deliberately colocated because they MUST be fed the identical inputs — the normalized +/// `url::Url::as_str()` serialization and the expected digest. If what goes into this key ever +/// changes (as #21215 changed it), bump the format-version salt in `MARKER_ARGUMENTS` so the +/// remote tier changes with it rather than silently desynchronizing. +pub(crate) fn observed_url_key(url: &Url, digest: Digest) -> CacheKey { + let observed_url = ObservedUrl { + url: url.as_str().to_owned(), + observed_digest: Some(digest.into()), + }; + CacheKey { + key_type: CacheKeyType::Url.into(), + digest: Some(Digest::of_bytes(&observed_url.to_bytes()).into()), + } +} + +/// A synthetic, never-executed Command deterministically encoding the (URL, digest) pair: the +/// remote tier of `observed_url_key` above, fed the same inputs. See its doc comment. +fn make_marker_command(url: &Url, digest: Digest) -> remexec::Command { + let mut arguments: Vec = MARKER_ARGUMENTS.iter().map(ToString::to_string).collect(); + arguments.extend([ + url.as_str().to_owned(), + digest.hash.to_hex(), + digest.size_bytes.to_string(), + ]); + remexec::Command { + arguments, + output_paths: vec![MARKER_OUTPUT_PATH.to_owned()], + ..remexec::Command::default() + } +} + +/// The Action wrapping the marker Command. Its input root is the empty directory (whose +/// serialized `Directory` proto is the empty blob). +fn make_marker_action(command: &remexec::Command) -> remexec::Action { + remexec::Action { + command_digest: Some(Digest::of_bytes(&command.to_bytes()).into()), + input_root_digest: Some(hashing::EMPTY_DIGEST.into()), + ..remexec::Action::default() + } +} + +/// The ActionResult recording the verified association. Referencing the file as an output means +/// completeness-checking servers only return an AC hit while the blob is still present. +/// `is_executable` mirrors the download node's `snapshot_of_one_file(path, digest, true)`. +fn make_marker_action_result(digest: Digest) -> remexec::ActionResult { + remexec::ActionResult { + exit_code: 0, + output_files: vec![remexec::OutputFile { + path: MARKER_OUTPUT_PATH.to_owned(), + digest: Some(digest.into()), + is_executable: true, + ..remexec::OutputFile::default() + }], + ..remexec::ActionResult::default() + } +} + +/// The reader never trusts the ActionResult payload: it is solely an existence proof for the +/// digest the caller already expected. A payload whose output digest differs from the expected +/// digest is a cache miss — never an error, and never materialized. (A payload that fails to +/// decode at all instead surfaces as a provider error — kept as an error because the process +/// remote cache relies on that corruption signal — which the caller logs, throttled by +/// `remote_cache_warnings`, and likewise treats as a miss: either way the download falls back +/// to the origin and nothing is materialized.) +fn marker_matches(action_result: &remexec::ActionResult, expected_digest: Digest) -> bool { + if action_result.exit_code != 0 { + return false; + } + let [output_file] = action_result.output_files.as_slice() else { + return false; + }; + output_file.path == MARKER_OUTPUT_PATH + && require_digest(output_file.digest.as_ref()) == Ok(expected_digest) +} + +/// +/// Remote caching for `DownloadedFile` nodes, layered behind the local ObservedUrls cache. +/// +/// NB: This deliberately holds the full, remote-capable `Store` even in configurations where the +/// rest of the engine sees a local-only store: like the remote cache `CommandRunner`, the +/// download node is a remote cache code path. +/// +pub struct RemoteDownloadCache { + provider: Arc, + store: Store, + cache_read: bool, + cache_write: bool, + error_throttle: CacheErrorThrottle, + executor: Executor, +} + +impl RemoteDownloadCache { + pub fn new( + provider: Arc, + store: Store, + cache_read: bool, + cache_write: bool, + warnings_behavior: RemoteCacheWarningsBehavior, + executor: Executor, + ) -> Self { + Self { + provider, + store, + cache_read, + cache_write, + error_throttle: CacheErrorThrottle::new(warnings_behavior), + executor, + } + } + + /// + /// Attempt to serve the download from the remote cache: on an AC marker hit, fully + /// materialize the expected digest into the local store. Returns true if the download was + /// served. Errors are logged and treated as misses, so the caller falls back to the origin. + /// + /// URLs which are not cacheable remotely (see `url_is_cacheable_remotely`) are always a + /// miss: the check is enforced here, not (only) at call sites, because it is what keeps + /// secret-bearing URLs out of the shared remote cache. + /// + pub async fn load_cached_download(&self, url: &Url, digest: Digest, build_id: &str) -> bool { + if !self.cache_read || !url_is_cacheable_remotely(url) { + return false; + } + in_workunit!( + "remote_download_cache_read", + Level::Debug, + desc = Some(format!("Remote cache lookup for download: {url}")), + |workunit| async move { + workunit.increment_counter(Metric::RemoteDownloadCacheRequests, 1); + // Exactly one of the three outcome counters is incremented per request, so + // Requests == Cached + Uncached + ReadErrors, as for `RemoteCacheRequests*`. + let counter = match self.load_cached_download_inner(url, digest, build_id).await { + Ok(true) => { + log::debug!("remote download cache hit for: {url}"); + Metric::RemoteDownloadCacheRequestsCached + } + Ok(false) => { + log::debug!("remote download cache miss for: {url}"); + Metric::RemoteDownloadCacheRequestsUncached + } + Err(err) => { + self.error_throttle.log( + CacheErrorType::ReadError, + &format!("remote cache for download of {url}"), + err, + ); + Metric::RemoteDownloadCacheReadErrors + } + }; + workunit.increment_counter(counter, 1); + counter == Metric::RemoteDownloadCacheRequestsCached + } + ) + .await + } + + async fn load_cached_download_inner( + &self, + url: &Url, + digest: Digest, + build_id: &str, + ) -> Result { + let command = make_marker_command(url, digest); + let action = make_marker_action(&command); + let action_digest = Digest::of_bytes(&action.to_bytes()); + + let Some(action_result) = self + .provider + .get_action_result(action_digest, build_id) + .await? + else { + return Ok(false); + }; + if !marker_matches(&action_result, digest) { + log::debug!("Ignoring malformed remote download cache entry for {url}"); + return Ok(false); + } + + // Eagerly materialize the bytes into the local store: backtracking cannot rescue a + // download, so a `MissingDigest` surfacing later from its snapshot would be a hard + // failure. The content is digest-verified as it is fetched. + match self + .store + .ensure_downloaded(HashSet::from([digest]), HashSet::new()) + .await + { + Ok(()) => Ok(true), + // The marker outlived the file content (e.g. the blob was evicted from the remote + // store): a miss, not an error. The caller re-downloads from the origin and, when + // write-enabled, restores the marker/blob pairing. + Err(StoreError::MissingDigest(_, _)) => { + log::debug!( + "remote download cache entry for {url} was present, but its file content was \ + not; falling back to the origin" + ); + Ok(false) + } + Err(StoreError::Unclassified(err)) => Err(err), + } + } + + /// + /// Record the verified (URL, digest) association in the remote cache: upload the file bytes + /// and the synthetic Action/Command (and empty input root) protos, then write the AC marker. + /// + /// Writing the marker asserts that some machine actually fetched this URL and got these + /// bytes: it must only be called after the local store holds the digest-verified content. + /// + async fn write_back(&self, url: &Url, digest: Digest) -> Result<(), String> { + let command = make_marker_command(url, digest); + let action = make_marker_action(&command); + + let (command_digest, action_digest) = + remote::remote::ensure_action_stored_locally(&self.store, &command, &action).await?; + let input_root_digest = self + .store + .record_directory(&remexec::Directory::default(), true) + .await?; + + self.store + .ensure_remote_has_recursive(vec![ + digest, + command_digest, + action_digest, + input_root_digest, + ]) + .await + .map_err(|err| err.to_string())?; + + self.provider + .update_action_result(action_digest, make_marker_action_result(digest)) + .await + } + + /// + /// Spawn `write_back` on the session's tail tasks (as remote cache writes for processes are), + /// so downloads don't block on a multi-MiB upload. A short-lived run can exit before the + /// upload finishes and drop the write: this self-heals on a later cold, write-enabled run. + /// + /// URLs which are not cacheable remotely are never written: as for reads, the check is + /// enforced here because it is what keeps secret-bearing URLs out of the shared remote cache. + /// + pub fn spawn_write_back(self: &Arc, tail_tasks: TailTasks, url: Url, digest: Digest) { + if !self.cache_write || !url_is_cacheable_remotely(&url) { + return; + } + let this = self.clone(); + let task_name = format!("remote download cache write for {url}"); + let write_fut = in_workunit!("remote_download_cache_write", Level::Trace, |workunit| { + async move { + workunit.increment_counter(Metric::RemoteDownloadCacheWriteAttempts, 1); + match this.write_back(&url, digest).await { + Ok(()) => { + log::debug!("remote download cache updated for: {url}"); + workunit.increment_counter(Metric::RemoteDownloadCacheWriteSuccesses, 1); + } + Err(err) => { + this.error_throttle.log( + CacheErrorType::WriteError, + &format!("remote cache for download of {url}"), + err, + ); + workunit.increment_counter(Metric::RemoteDownloadCacheWriteErrors, 1); + } + } + } + }); + tail_tasks.spawn_on(&task_name, self.executor.handle(), write_fut.boxed()); + } +} + +/// Shared fixtures for this module's tests and the `DownloadedFile` node tests: one home for +/// the `RemoteStoreOptions` literal and the store + provider + cache assembly, so a field or +/// signature change is a single edit. +#[cfg(test)] +pub(crate) mod test_util { + use std::collections::BTreeMap; + use std::sync::Arc; + use std::time::Duration; + + use grpc_util::tls; + use store::{RemoteProvider, RemoteStoreOptions, Store}; + use tempfile::TempDir; + + use super::RemoteDownloadCache; + use remote::remote_cache::RemoteCacheWarningsBehavior; + + pub(crate) fn remote_store_options( + provider: RemoteProvider, + address: String, + ) -> RemoteStoreOptions { + RemoteStoreOptions { + provider, + store_address: address, + instance_name: None, + tls_config: tls::Config::default(), + headers: BTreeMap::new(), + chunk_size_bytes: 10 * 1024 * 1024, + timeout: Duration::from_secs(5), + retries: 0, + concurrency_limit: 256, + batch_api_size_limit: 4 * 1024 * 1024, + batch_load_enabled: false, + } + } + + /// A fresh local store (in its own TempDir) with `options`' remote attached, plus a + /// `RemoteDownloadCache` holding that same (full, remote-capable) store. + pub(crate) async fn make_download_cache( + options: RemoteStoreOptions, + cache_read: bool, + cache_write: bool, + ) -> (TempDir, Store, Arc) { + let executor = task_executor::Executor::new(); + let dir = TempDir::new().unwrap(); + let store = Store::local_only(executor.clone(), dir.path()) + .unwrap() + .into_with_remote(options.clone()) + .await + .unwrap(); + let provider = remote_provider::choose_action_cache_provider(options) + .await + .unwrap(); + let cache = Arc::new(RemoteDownloadCache::new( + provider, + store.clone(), + cache_read, + cache_write, + RemoteCacheWarningsBehavior::FirstOnly, + executor, + )); + (dir, store, cache) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::{Duration, Instant}; + + use bytes::Bytes; + use hashing::Digest; + use store::{RemoteProvider, Store}; + use task_executor::TailTasks; + use tempfile::TempDir; + use testutil_mock::{RequestType, StubCAS}; + use url::Url; + use workunit_store::WorkunitStore; + + use super::test_util::{make_download_cache, remote_store_options}; + use super::*; + + const TEST_URL: &str = "https://example.com/tool.tar.gz"; + const TEST_BYTES: &[u8] = b"downloaded tool bytes"; + + fn test_url() -> Url { + Url::parse(TEST_URL).unwrap() + } + + fn test_digest() -> Digest { + Digest::of_bytes(TEST_BYTES) + } + + fn marker_action_digest(url: &Url, digest: Digest) -> Digest { + Digest::of_bytes(&make_marker_action(&make_marker_command(url, digest)).to_bytes()) + } + + async fn make_cache( + cas: &StubCAS, + cache_read: bool, + cache_write: bool, + ) -> (TempDir, Store, Arc) { + make_download_cache( + remote_store_options(RemoteProvider::Reapi, cas.address()), + cache_read, + cache_write, + ) + .await + } + + #[test] + fn url_eligibility() { + for eligible in [ + "https://example.com/foo", + "http://example.com/foo?presigned=abc", + ] { + assert!(url_is_cacheable_remotely(&Url::parse(eligible).unwrap())); + } + for ineligible in [ + "file:/tmp/foo", + "https://user:token@example.com/foo", + "https://user@example.com/foo", + "https://:token@example.com/foo", + ] { + assert!(!url_is_cacheable_remotely(&Url::parse(ineligible).unwrap())); + } + } + + #[test] + fn marker_encoding() { + let url = test_url(); + let digest = test_digest(); + + let command = make_marker_command(&url, digest); + assert_eq!( + command.arguments, + vec![ + "__pants_url_download__".to_owned(), + "v1".to_owned(), + TEST_URL.to_owned(), + digest.hash.to_hex(), + TEST_BYTES.len().to_string(), + ] + ); + assert_eq!(command.output_paths, vec!["file".to_owned()]); + + let action = make_marker_action(&command); + assert_eq!( + action.command_digest, + Some(Digest::of_bytes(&command.to_bytes()).into()) + ); + assert_eq!(action.input_root_digest, Some(hashing::EMPTY_DIGEST.into())); + + // Distinct URLs and distinct digests must key distinct actions. + let other_url = Url::parse("https://example.com/other.tar.gz").unwrap(); + assert_ne!( + marker_action_digest(&url, digest), + marker_action_digest(&other_url, digest) + ); + assert_ne!( + marker_action_digest(&url, digest), + marker_action_digest(&url, Digest::of_bytes(b"other content")) + ); + } + + #[test] + fn marker_validation() { + let digest = test_digest(); + + assert!(marker_matches(&make_marker_action_result(digest), digest)); + + let wrong_digest = make_marker_action_result(Digest::of_bytes(b"other content")); + assert!(!marker_matches(&wrong_digest, digest)); + + assert!(!marker_matches(&remexec::ActionResult::default(), digest)); + + let mut failed = make_marker_action_result(digest); + failed.exit_code = 1; + assert!(!marker_matches(&failed, digest)); + + let mut wrong_path = make_marker_action_result(digest); + wrong_path.output_files[0].path = "other".to_owned(); + assert!(!marker_matches(&wrong_path, digest)); + + let mut extra_output = make_marker_action_result(digest); + extra_output + .output_files + .push(extra_output.output_files[0].clone()); + assert!(!marker_matches(&extra_output, digest)); + + let mut no_digest = make_marker_action_result(digest); + no_digest.output_files[0].digest = None; + assert!(!marker_matches(&no_digest, digest)); + } + + #[tokio::test] + async fn write_back_then_cold_read() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let url = test_url(); + let digest = test_digest(); + + // A write-enabled machine holds the digest-verified download locally, and writes back. + let (_dir_a, store_a, cache_a) = make_cache(&cas, true, true).await; + store_a + .store_file_bytes(Bytes::from_static(TEST_BYTES), true) + .await + .unwrap(); + cache_a.write_back(&url, digest).await.unwrap(); + + assert!(cas.contains(digest.hash)); + assert!(cas.contains_action_result(marker_action_digest(&url, digest).hash)); + + // A cold machine is served entirely from the remote cache, with the bytes fully + // materialized into its local store. + let (_dir_b, store_b, cache_b) = make_cache(&cas, true, true).await; + assert!(cache_b.load_cached_download(&url, digest, "build_id").await); + let loaded = store_b + .clone() + .into_local_only() + .load_file_bytes_with(digest, Bytes::copy_from_slice) + .await + .unwrap(); + assert_eq!(loaded, Bytes::from_static(TEST_BYTES)); + } + + #[tokio::test] + async fn write_back_then_cold_read_with_file_provider() { + // The same round trip as `write_back_then_cold_read`, against the OpenDAL + // `experimental-file` provider: download caching with zero REAPI infrastructure. + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let remote_dir = TempDir::new().unwrap(); + let url = test_url(); + let digest = test_digest(); + + let remote_options = remote_store_options( + RemoteProvider::ExperimentalFile, + format!("file://{}", remote_dir.path().display()), + ); + let make_cache = async |cache_read: bool, cache_write: bool| { + make_download_cache(remote_options.clone(), cache_read, cache_write).await + }; + + let (_dir_a, store_a, cache_a) = make_cache(true, true).await; + store_a + .store_file_bytes(Bytes::from_static(TEST_BYTES), true) + .await + .unwrap(); + cache_a.write_back(&url, digest).await.unwrap(); + + let (_dir_b, store_b, cache_b) = make_cache(true, true).await; + assert!(cache_b.load_cached_download(&url, digest, "build_id").await); + let loaded = store_b + .clone() + .into_local_only() + .load_file_bytes_with(digest, Bytes::copy_from_slice) + .await + .unwrap(); + assert_eq!(loaded, Bytes::from_static(TEST_BYTES)); + } + + #[tokio::test] + async fn blob_without_marker_is_a_miss() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let digest = test_digest(); + // The digest is already in the CAS (e.g. as a process output), but no machine has + // verified that this URL serves it: strict semantics require a real download. + let cas = StubCAS::builder() + .unverified_content(digest.hash, Bytes::from_static(TEST_BYTES)) + .build() + .await; + + let (_dir, _store, cache) = make_cache(&cas, true, true).await; + assert!( + !cache + .load_cached_download(&test_url(), digest, "build_id") + .await + ); + // The miss came from actually consulting the AC, not from skipping the read path. + assert_eq!(cas.request_count(RequestType::ACGetActionResult), 1); + } + + #[tokio::test] + async fn ineligible_urls_make_no_requests() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let digest = test_digest(); + + // Even with read and write enabled and the bytes locally present, `file:` and + // userinfo-bearing URLs must never reach the remote cache — enforced by the methods + // themselves, independent of any call-site filtering. + let (_dir, store, cache) = make_cache(&cas, true, true).await; + store + .store_file_bytes(Bytes::from_static(TEST_BYTES), true) + .await + .unwrap(); + + for ineligible in [ + "file:/tmp/tool.tar.gz", + "https://user:token@example.com/tool.tar.gz", + ] { + let url = Url::parse(ineligible).unwrap(); + assert!(!cache.load_cached_download(&url, digest, "build_id").await); + let tail_tasks = TailTasks::new(); + cache.spawn_write_back(tail_tasks.clone(), url, digest); + tail_tasks.wait(Duration::from_secs(10)).await; + } + + assert_eq!(cas.request_count(RequestType::ACGetActionResult), 0); + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 0); + assert!(!cas.contains(digest.hash)); + } + + #[tokio::test] + async fn marker_with_evicted_blob_is_a_miss() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let url = test_url(); + let digest = test_digest(); + + let (_dir_a, store_a, cache_a) = make_cache(&cas, true, true).await; + store_a + .store_file_bytes(Bytes::from_static(TEST_BYTES), true) + .await + .unwrap(); + cache_a.write_back(&url, digest).await.unwrap(); + assert!(cas.remove(digest.hash)); + + let (_dir_b, _store_b, cache_b) = make_cache(&cas, true, true).await; + assert!(!cache_b.load_cached_download(&url, digest, "build_id").await); + } + + #[tokio::test] + async fn malformed_marker_is_a_miss() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let url = test_url(); + let digest = test_digest(); + let action_digest = marker_action_digest(&url, digest); + let (_dir, _store, cache) = make_cache(&cas, true, true).await; + + // A marker with no output files at all. + cas.action_cache.insert( + action_digest, + 0, + hashing::EMPTY_DIGEST, + hashing::EMPTY_DIGEST, + ); + assert!(!cache.load_cached_download(&url, digest, "build_id").await); + + // A marker whose payload names a different digest than the expected one: it must never + // be materialized. + let poisoned = make_marker_action_result(Digest::of_bytes(b"attacker controlled")); + cas.action_cache + .action_map + .lock() + .insert(action_digest.hash, poisoned); + assert!(!cache.load_cached_download(&url, digest, "build_id").await); + } + + #[tokio::test] + async fn ac_errors_are_misses() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::builder().ac_always_errors().build().await; + let (_dir, _store, cache) = make_cache(&cas, true, true).await; + assert!( + !cache + .load_cached_download(&test_url(), test_digest(), "build_id") + .await + ); + } + + #[tokio::test] + async fn hung_cache_lookups_time_out_and_are_misses() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + // A degraded cache server which accepts the request and then hangs (e.g. a blackholed + // host), for far longer than the client's configured RPC timeout. + let cas = StubCAS::builder() + .ac_read_delay(Duration::from_secs(60)) + .build() + .await; + let mut options = remote_store_options(RemoteProvider::Reapi, cas.address()); + options.timeout = Duration::from_millis(250); + let (_dir, _store, cache) = make_download_cache(options, true, true).await; + + let start = Instant::now(); + assert!( + !cache + .load_cached_download(&test_url(), test_digest(), "build_id") + .await + ); + // The lookup gave up within the configured budget ((retries + 1) x the RPC timeout) and + // became a miss — so the caller falls back to the origin — rather than waiting on the + // server's response or failing the build. + assert!(start.elapsed() < Duration::from_secs(10)); + } + + #[tokio::test] + async fn read_disabled_makes_no_requests() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let (_dir, _store, cache) = make_cache(&cas, false, true).await; + assert!( + !cache + .load_cached_download(&test_url(), test_digest(), "build_id") + .await + ); + assert_eq!(cas.request_count(RequestType::ACGetActionResult), 0); + } + + #[tokio::test] + async fn spawned_write_back_writes() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let url = test_url(); + let digest = test_digest(); + + let (_dir, store, cache) = make_cache(&cas, true, true).await; + store + .store_file_bytes(Bytes::from_static(TEST_BYTES), true) + .await + .unwrap(); + let tail_tasks = TailTasks::new(); + cache.spawn_write_back(tail_tasks.clone(), url.clone(), digest); + tail_tasks.wait(Duration::from_secs(10)).await; + + assert!(cas.contains(digest.hash)); + assert!(cas.contains_action_result(marker_action_digest(&url, digest).hash)); + } + + #[tokio::test] + async fn write_disabled_makes_no_requests() { + let (_workunit_store, _workunit) = WorkunitStore::setup_for_tests(); + let cas = StubCAS::empty().await; + let digest = test_digest(); + + let (_dir, store, cache) = make_cache(&cas, true, false).await; + store + .store_file_bytes(Bytes::from_static(TEST_BYTES), true) + .await + .unwrap(); + let tail_tasks = TailTasks::new(); + cache.spawn_write_back(tail_tasks.clone(), test_url(), digest); + tail_tasks.wait(Duration::from_secs(10)).await; + + assert_eq!(cas.request_count(RequestType::ACUpdateActionResult), 0); + assert!(!cas.contains(digest.hash)); + } +} diff --git a/src/rust/remote_provider/remote_provider_opendal/src/action_cache_tests.rs b/src/rust/remote_provider/remote_provider_opendal/src/action_cache_tests.rs index 063926aa6c0..4968933d9db 100644 --- a/src/rust/remote_provider/remote_provider_opendal/src/action_cache_tests.rs +++ b/src/rust/remote_provider/remote_provider_opendal/src/action_cache_tests.rs @@ -90,6 +90,26 @@ async fn get_action_result_missing() { ); } +#[tokio::test] +async fn get_action_result_undecodable_is_an_error() { + let provider = new_provider(); + + let action_digest = Digest::of_bytes(b"get_action_result_undecodable test"); + // A corrupt/truncated entry, e.g. from an interrupted write on a backend without atomic + // writes. This provider loads action results without validation, so the garbage reaches the + // decoder, and must surface as an error (not a silent miss): consumers rely on it for + // corruption logging and metrics, and fall back as if it were a miss regardless. + provider + .operator + .write(&test_path(action_digest), &b"\xff\xffnot a valid proto"[..]) + .await + .unwrap(); + + let result = provider.get_action_result(action_digest, "").await; + let err = result.expect_err("undecodable action result should be an error"); + assert!(err.contains("failed to decode action result"), "{err}"); +} + #[tokio::test] async fn update_action_cache() { let provider = new_provider(); diff --git a/src/rust/remote_provider/remote_provider_opendal/src/lib.rs b/src/rust/remote_provider/remote_provider_opendal/src/lib.rs index 8126aad4da6..ea7d2a72665 100644 --- a/src/rust/remote_provider/remote_provider_opendal/src/lib.rs +++ b/src/rust/remote_provider/remote_provider_opendal/src/lib.rs @@ -334,6 +334,10 @@ impl ActionCacheProvider for Provider { false => Ok(None), true => { let bytes = Bytes::from(destination); + // NB: A corrupt or truncated entry (e.g. from an interrupted write on a backend + // without atomic writes) is deliberately an error, not a silent miss: consumers + // (the process remote cache, the remote download cache) treat read errors as + // misses anyway, but rely on the error for their corruption logging and metrics. Ok(Some(ActionResult::decode(bytes).map_err(|e| { format!("failed to decode action result for digest {action_digest:?}: {e}") })?)) diff --git a/src/rust/workunit_store/src/metrics.rs b/src/rust/workunit_store/src/metrics.rs index 748618c8713..b22806d8a3b 100644 --- a/src/rust/workunit_store/src/metrics.rs +++ b/src/rust/workunit_store/src/metrics.rs @@ -41,6 +41,18 @@ pub enum Metric { /// processes directly. RemoteCacheTotalTimeSavedMs, RemoteCacheRequestTimeouts, + /// Number of lookups of a URL download in the remote cache. + RemoteDownloadCacheRequests, + /// Number of URL downloads served entirely from the remote cache. + RemoteDownloadCacheRequestsCached, + /// Number of URL downloads which were looked up in the remote cache, but which were then + /// fetched from their origin URL (because the cache entry was absent, invalid, or its file + /// content had been evicted). + RemoteDownloadCacheRequestsUncached, + RemoteDownloadCacheReadErrors, + RemoteDownloadCacheWriteAttempts, + RemoteDownloadCacheWriteSuccesses, + RemoteDownloadCacheWriteErrors, RemoteExecutionErrors, RemoteExecutionRequests, RemoteExecutionRPCErrors,