diff --git a/BUILD.bazel b/BUILD.bazel index f9a482dea..700ba5dd3 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -133,3 +133,12 @@ sh_test( timeout = "short", srcs = [":dummy_test_sh"], ) + +# Large deterministic output used by integration_tests/chunking_cache_test.sh +# to exercise --experimental_remote_cache_chunking (must exceed the chunking +# threshold of 4x the average chunk size). +genrule( + name = "chunking_test_artifact", + outs = ["chunking_test_artifact.txt"], + cmd = "seq 1 1000000 > \"$@\"", +) diff --git a/Cargo.lock b/Cargo.lock index 85647be66..9efaf70f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1618,6 +1618,17 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "fastcdc" +version = "3.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf51ceb43e96afbfe4dd5c6f6082af5dfd60e220820b8123792d61963f2ce6bc" +dependencies = [ + "async-stream", + "tokio", + "tokio-stream", +] + [[package]] name = "fastrand" version = "2.4.1" @@ -3124,6 +3135,7 @@ dependencies = [ "async-trait", "axum", "bytes", + "fastcdc", "futures", "hex", "http-body-util", @@ -3149,6 +3161,7 @@ dependencies = [ "sha2", "tokio", "tokio-stream", + "tokio-util", "tonic", "tonic-prost", "tower", diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index 52ac7b662..89d6e6a81 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -931,6 +931,7 @@ "errno_0.3.14": "{\"dependencies\":[{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2\",\"target\":\"cfg(target_os=\\\"hermit\\\")\"},{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2\",\"target\":\"cfg(target_os=\\\"wasi\\\")\"},{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2\",\"target\":\"cfg(unix)\"},{\"features\":[\"Win32_Foundation\",\"Win32_System_Diagnostics_Debug\"],\"name\":\"windows-sys\",\"req\":\">=0.52, <0.62\",\"target\":\"cfg(windows)\"}],\"features\":{\"default\":[\"std\"],\"std\":[\"libc/std\"]}}", "event-listener-strategy_0.5.4": "{\"dependencies\":[{\"default_features\":false,\"name\":\"event-listener\",\"req\":\"^5.0.0\"},{\"kind\":\"dev\",\"name\":\"futures-lite\",\"req\":\"^2.0.0\"},{\"name\":\"pin-project-lite\",\"req\":\"^0.2.12\"},{\"kind\":\"dev\",\"name\":\"wasm-bindgen-test\",\"req\":\"^0.3.37\",\"target\":\"cfg(target_family = \\\"wasm\\\")\"}],\"features\":{\"default\":[\"std\"],\"loom\":[\"event-listener/loom\"],\"portable-atomic\":[\"event-listener/portable-atomic\"],\"std\":[\"event-listener/std\"]}}", "event-listener_5.4.1": "{\"dependencies\":[{\"default_features\":false,\"name\":\"concurrent-queue\",\"req\":\"^2.4.0\"},{\"default_features\":false,\"features\":[\"cargo_bench_support\"],\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.7\"},{\"default_features\":false,\"name\":\"critical-section\",\"optional\":true,\"req\":\"^1.2.0\"},{\"features\":[\"std\"],\"kind\":\"dev\",\"name\":\"critical-section\",\"req\":\"^1.2.0\"},{\"kind\":\"dev\",\"name\":\"futures-lite\",\"req\":\"^2.0.0\"},{\"name\":\"loom\",\"optional\":true,\"req\":\"^0.7\",\"target\":\"cfg(loom)\"},{\"name\":\"parking\",\"optional\":true,\"req\":\"^2.0.0\",\"target\":\"cfg(not(target_family = \\\"wasm\\\"))\"},{\"name\":\"pin-project-lite\",\"req\":\"^0.2.12\"},{\"default_features\":false,\"features\":[\"alloc\"],\"name\":\"portable-atomic-util\",\"optional\":true,\"req\":\"^0.2.0\"},{\"default_features\":false,\"name\":\"portable_atomic_crate\",\"optional\":true,\"package\":\"portable-atomic\",\"req\":\"^1.2.0\"},{\"kind\":\"dev\",\"name\":\"try-lock\",\"req\":\"^0.2.5\"},{\"kind\":\"dev\",\"name\":\"waker-fn\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"wasm-bindgen-test\",\"req\":\"^0.3\",\"target\":\"cfg(target_family = \\\"wasm\\\")\"}],\"features\":{\"default\":[\"std\"],\"loom\":[\"concurrent-queue/loom\",\"parking?/loom\",\"dep:loom\"],\"portable-atomic\":[\"portable-atomic-util\",\"portable_atomic_crate\",\"concurrent-queue/portable-atomic\"],\"std\":[\"concurrent-queue/std\",\"parking\"]}}", + "fastcdc_3.2.1": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"aes\",\"req\":\"^0.8.2\"},{\"name\":\"async-stream\",\"optional\":true,\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"byteorder\",\"req\":\"^1.4.3\"},{\"features\":[\"cargo\"],\"kind\":\"dev\",\"name\":\"clap\",\"req\":\"^4.2.1\"},{\"kind\":\"dev\",\"name\":\"ctr\",\"req\":\"^0.9.2\"},{\"name\":\"futures\",\"optional\":true,\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"futures-test\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"md-5\",\"req\":\"^0.10.5\"},{\"kind\":\"dev\",\"name\":\"memmap2\",\"req\":\"^0.9.5\"},{\"features\":[\"io-util\"],\"name\":\"tokio\",\"optional\":true,\"req\":\"^1\"},{\"features\":[\"fs\",\"io-util\",\"rt\",\"rt-multi-thread\",\"macros\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"name\":\"tokio-stream\",\"optional\":true,\"req\":\"^0.1\"}],\"features\":{\"default\":[],\"futures\":[\"dep:futures\"],\"tokio\":[\"dep:tokio\",\"tokio-stream\",\"async-stream\"]}}", "fastrand_2.3.0": "{\"dependencies\":[{\"features\":[\"js\"],\"name\":\"getrandom\",\"optional\":true,\"req\":\"^0.2\",\"target\":\"cfg(all(any(target_arch = \\\"wasm32\\\", target_arch = \\\"wasm64\\\"), target_os = \\\"unknown\\\"))\"},{\"kind\":\"dev\",\"name\":\"getrandom\",\"req\":\"^0.2\"},{\"features\":[\"js\"],\"kind\":\"dev\",\"name\":\"getrandom\",\"req\":\"^0.2\",\"target\":\"cfg(all(any(target_arch = \\\"wasm32\\\", target_arch = \\\"wasm64\\\"), target_os = \\\"unknown\\\"))\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8\"},{\"kind\":\"dev\",\"name\":\"wasm-bindgen-test\",\"req\":\"^0.3\",\"target\":\"cfg(all(any(target_arch = \\\"wasm32\\\", target_arch = \\\"wasm64\\\"), target_os = \\\"unknown\\\"))\"},{\"kind\":\"dev\",\"name\":\"wyhash\",\"req\":\"^0.5\"}],\"features\":{\"alloc\":[],\"default\":[\"std\"],\"js\":[\"std\",\"getrandom\"],\"std\":[\"alloc\"]}}", "fastrand_2.4.1": "{\"dependencies\":[{\"features\":[\"wasm_js\"],\"name\":\"getrandom\",\"optional\":true,\"req\":\"^0.3.4\",\"target\":\"cfg(all(any(target_arch = \\\"wasm32\\\", target_arch = \\\"wasm64\\\"), target_os = \\\"unknown\\\"))\"},{\"kind\":\"dev\",\"name\":\"getrandom\",\"req\":\"^0.3.4\"},{\"features\":[\"wasm_js\"],\"kind\":\"dev\",\"name\":\"getrandom\",\"req\":\"^0.3.4\",\"target\":\"cfg(all(any(target_arch = \\\"wasm32\\\", target_arch = \\\"wasm64\\\"), target_os = \\\"unknown\\\"))\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.9\"},{\"kind\":\"dev\",\"name\":\"wasm-bindgen-test\",\"req\":\"^0.3\",\"target\":\"cfg(all(any(target_arch = \\\"wasm32\\\", target_arch = \\\"wasm64\\\"), target_os = \\\"unknown\\\"))\"},{\"kind\":\"dev\",\"name\":\"wyhash\",\"req\":\"^0.6\"}],\"features\":{\"alloc\":[],\"default\":[\"std\"],\"js\":[\"std\",\"getrandom\"],\"std\":[\"alloc\"]}}", "ff_0.13.1": "{\"dependencies\":[{\"default_features\":false,\"name\":\"bitvec\",\"optional\":true,\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"blake2b_simd\",\"req\":\"^1\"},{\"default_features\":false,\"name\":\"byteorder\",\"optional\":true,\"req\":\"^1\"},{\"name\":\"ff_derive\",\"optional\":true,\"req\":\"^0.13.1\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8\"},{\"default_features\":false,\"name\":\"rand_core\",\"req\":\"^0.6\"},{\"default_features\":false,\"features\":[\"i128\"],\"name\":\"subtle\",\"req\":\"^2.2.1\"}],\"features\":{\"alloc\":[],\"bits\":[\"bitvec\"],\"default\":[\"bits\",\"std\"],\"derive\":[\"byteorder\",\"ff_derive\"],\"derive_bits\":[\"bits\",\"ff_derive/bits\"],\"std\":[\"alloc\"]}}", diff --git a/deployment-examples/docker-compose/local-storage-cas.json5 b/deployment-examples/docker-compose/local-storage-cas.json5 index 6d4acdeed..69aae0134 100644 --- a/deployment-examples/docker-compose/local-storage-cas.json5 +++ b/deployment-examples/docker-compose/local-storage-cas.json5 @@ -13,8 +13,8 @@ }, backend: { filesystem: { - content_path: "~/.cache/nativelink/content_path-cas", - temp_path: "~/.cache/nativelink/tmp_path-cas", + content_path: "/root/.cache/nativelink/content_path-cas", + temp_path: "/root/.cache/nativelink/tmp_path-cas", eviction_policy: { // 10gb. max_bytes: 10000000000, @@ -23,11 +23,25 @@ }, }, }, + { + // Holds blob-to-chunks layouts for the SplitBlob/SpliceBlob RPCs used + // by Bazel's --experimental_remote_cache_chunking. Must not verify + // digests and must not be the same store as the CAS. + name: "CHUNK_INDEX_STORE", + filesystem: { + content_path: "/root/.cache/nativelink/content_path-chunk-index", + temp_path: "/root/.cache/nativelink/tmp_path-chunk-index", + eviction_policy: { + // 100mb. + max_bytes: 100000000, + }, + }, + }, { name: "AC_MAIN_STORE", filesystem: { - content_path: "~/.cache/nativelink/content_path-ac", - temp_path: "~/.cache/nativelink/tmp_path-ac", + content_path: "/root/.cache/nativelink/content_path-ac", + temp_path: "/root/.cache/nativelink/tmp_path-ac", eviction_policy: { // 500mb. max_bytes: 500000000, @@ -46,6 +60,13 @@ cas: [ { cas_store: "CAS_MAIN_STORE", + + // Optional: enables content-defined chunking + // (SplitBlob/SpliceBlob) for Bazel clients running with + // --experimental_remote_cache_chunking. + experimental_chunking: { + index_store: "CHUNK_INDEX_STORE", + }, }, ], ac: [ diff --git a/integration_tests/chunking_cache_test.sh b/integration_tests/chunking_cache_test.sh new file mode 100755 index 000000000..16351c1c1 --- /dev/null +++ b/integration_tests/chunking_cache_test.sh @@ -0,0 +1,74 @@ +#!/bin/bash +# Copyright 2026 The NativeLink Authors. All rights reserved. +# +# Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# See LICENSE file for details +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Sanity check for REAPI content-defined chunking: uploads a large blob via +# SpliceBlob (Bazel --experimental_remote_cache_chunking), verifies the +# server registered a chunk layout, then re-fetches the blob from the remote +# cache via the chunked download path and checks it is byte-identical. + +if [[ $UNDER_TEST_RUNNER -ne 1 ]]; then + echo "This script should be run under run_integration_tests.sh" + exit 1 +fi +set -x + +# Bazel uploads cache entries in the background by default +# (--remote_cache_async, on since Bazel 8), so a build can return before the +# chunked upload and SpliceBlob complete. Force synchronous uploads so the +# chunk-index assertions below cannot race the upload. +CHUNKING_FLAGS=(--config self_test --experimental_remote_cache_chunking --remote_cache_async=false) +EXPECTED_SHA=$(seq 1 1000000 | sha256sum | awk '{print $1}') +# The test runner's working directory is not the workspace root, so resolve +# the output location through bazel itself. +ARTIFACT="$(bazel --output_base="$BAZEL_CACHE_DIR" info "${CHUNKING_FLAGS[@]}" bazel-bin)/chunking_test_artifact.txt" + +# First build executes the action locally and uploads the ~6.9MB output as +# chunks (SpliceBlob). +bazel --output_base="$BAZEL_CACHE_DIR" build "${CHUNKING_FLAGS[@]}" //:chunking_test_artifact +FIRST_SHA=$(sha256sum "$ARTIFACT" | awk '{print $1}') +if [[ $FIRST_SHA != "$EXPECTED_SHA" ]]; then + echo "Expected locally built artifact to have sha $EXPECTED_SHA, got $FIRST_SHA." + exit 1 +fi + +# The server must have registered a chunk layout for the spliced blob. The +# index store is mounted from the host by docker-compose. +CHUNK_INDEX_DIR="${NATIVELINK_DIR:-$HOME/.cache/nativelink}/content_path-chunk-index" +if [[ -z $(find "$CHUNK_INDEX_DIR" -type f 2> /dev/null) ]]; then + echo "Expected a chunk layout in $CHUNK_INDEX_DIR after a chunked upload." + echo "SpliceBlob was likely not used; check that the server advertises" + echo "chunking support and that bazel supports the chunking flag." + echo "Diagnostics: contents of the mounted cache dir and root's default:" + sudo find "${NATIVELINK_DIR:-$HOME/.cache/nativelink}" -maxdepth 1 2> /dev/null || true + sudo find /root/.cache/nativelink -maxdepth 1 2> /dev/null || true + exit 1 +fi + +# Clean our local cache and re-fetch from the remote cache through the +# chunked download path. +bazel --output_base="$BAZEL_CACHE_DIR" clean +OUTPUT=$(bazel --output_base="$BAZEL_CACHE_DIR" build "${CHUNKING_FLAGS[@]}" //:chunking_test_artifact 2>&1) +if [[ ! $OUTPUT =~ 'remote cache hit' ]]; then + echo "Expected second bazel run to be a remote cache hit." + echo "STDOUT:" + echo "$OUTPUT" + exit 1 +fi +SECOND_SHA=$(sha256sum "$ARTIFACT" | awk '{print $1}') +if [[ $SECOND_SHA != "$EXPECTED_SHA" ]]; then + echo "Artifact fetched through the chunked download path is corrupt:" + echo "expected sha $EXPECTED_SHA, got $SECOND_SHA." + exit 1 +fi diff --git a/nativelink-config/examples/chunking_cas.json5 b/nativelink-config/examples/chunking_cas.json5 new file mode 100644 index 000000000..f2ca4b966 --- /dev/null +++ b/nativelink-config/examples/chunking_cas.json5 @@ -0,0 +1,105 @@ +// Demonstrates REAPI content-defined chunking: the SplitBlob/SpliceBlob +// RPCs used by Bazel's --experimental_remote_cache_chunking flag +// (available in Bazel 8.7.0+ / 9.1.0+). +// +// Chunking is entirely optional and disabled by default: without the +// `experimental_chunking` block below, NativeLink behaves exactly as +// before and does not advertise chunking support. When enabled, clients +// upload and download large blobs as content-defined chunks, so small +// changes to large outputs only transfer the chunks that changed. +{ + stores: [ + { + name: "CAS_MAIN_STORE", + filesystem: { + content_path: "/tmp/nativelink/data/content_path-cas", + temp_path: "/tmp/nativelink/data/tmp_path-cas", + eviction_policy: { + // 10gb. + max_bytes: 10000000000, + }, + }, + }, + { + // Holds the blob-to-chunks layouts registered via SpliceBlob or + // created by on-demand chunking in SplitBlob. Layout entries are + // small (roughly 80-140 bytes per chunk). This store must not verify + // content digests and must not be the same store as the CAS itself. + name: "CHUNK_INDEX_STORE", + filesystem: { + content_path: "/tmp/nativelink/data/content_path-chunk-index", + temp_path: "/tmp/nativelink/data/tmp_path-chunk-index", + eviction_policy: { + // 100mb. + max_bytes: 100000000, + }, + }, + }, + { + name: "AC_MAIN_STORE", + filesystem: { + content_path: "/tmp/nativelink/data/content_path-ac", + temp_path: "/tmp/nativelink/data/tmp_path-ac", + eviction_policy: { + // 500mb. + max_bytes: 500000000, + }, + }, + }, + ], + servers: [ + { + listener: { + http: { + socket_address: "0.0.0.0:50051", + }, + }, + services: { + cas: [ + { + instance_name: "main", + cas_store: "CAS_MAIN_STORE", + + // Optional: omit this block to disable chunking entirely. + experimental_chunking: { + // Required, unless `cas_store` is a grpc store — in that + // case the chunking RPCs are forwarded to the backend and + // `index_store` must be omitted. + index_store: "CHUNK_INDEX_STORE", + + // Optional: the average chunk size in bytes advertised to + // clients and used for server-side chunking. Must be between + // 1 KiB and 1 MiB. + // Default: 524288 (512 KiB). + avg_chunk_size_bytes: 524288, + + // Optional: blobs that would produce more chunks than this + // are served without chunking. + // Default: 50000. + max_chunk_count: 50000, + }, + }, + ], + ac: [ + { + instance_name: "main", + ac_store: "AC_MAIN_STORE", + }, + ], + + // The capabilities service advertises chunking support; Bazel only + // issues SplitBlob/SpliceBlob when it is advertised. + capabilities: [ + { + instance_name: "main", + }, + ], + bytestream: { + cas_stores: { + main: "CAS_MAIN_STORE", + }, + }, + }, + }, + ], +} diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index 5015e9aea..5f8b72984 100644 --- a/nativelink-config/src/cas_server.rs +++ b/nativelink-config/src/cas_server.rs @@ -121,7 +121,7 @@ pub struct AcStoreConfig { pub read_only: bool, } -#[derive(Deserialize, Serialize, Debug)] +#[derive(Deserialize, Serialize, Debug, Clone)] #[serde(deny_unknown_fields)] #[cfg_attr(feature = "dev-schema", derive(JsonSchema))] pub struct CasStoreConfig { @@ -129,6 +129,115 @@ pub struct CasStoreConfig { /// This store name referenced here may be reused multiple times. #[serde(deserialize_with = "convert_string_with_shellexpand")] pub cas_store: StoreRefName, + + /// Optional and experimental: enables the REAPI `SplitBlob`/`SpliceBlob` + /// RPCs used by content-defined chunking clients (e.g. Bazel's + /// `--experimental_remote_cache_chunking`). When set, the capabilities + /// service advertises blob split/splice support and `FastCDC` 2020 + /// parameters for this instance. When `cas_store` is a grpc store the + /// RPCs are forwarded to the backend (which must support chunking with + /// matching parameters); otherwise they are served locally. + /// + /// See `nativelink-config/examples/chunking_cas.json5` for a complete + /// configuration example. + /// + /// Default: not set — chunking RPCs are rejected, nothing is advertised, + /// and behavior is identical to when this option did not exist. + #[serde(default)] + pub experimental_chunking: Option, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +#[serde(deny_unknown_fields)] +#[cfg_attr(feature = "dev-schema", derive(JsonSchema))] +pub struct CasChunkingConfig { + /// The store name referenced in the `stores` map in the main config used + /// to persist blob-to-chunks layouts. Keys are the digests of the + /// original blobs and values are serialized chunk layouts (which do not + /// hash to those digests), so this store MUST NOT perform content digest + /// verification and MUST NOT be the same store as `cas_store` — writing + /// layouts into the CAS would overwrite blob content. Using the same + /// store name as `cas_store` is rejected at startup. + /// + /// Required unless `cas_store` is a grpc store: for proxied instances + /// the `SplitBlob`/`SpliceBlob` RPCs are forwarded to the backend, which + /// owns the chunk layouts, and setting an `index_store` is rejected at + /// startup. + #[serde(default, deserialize_with = "convert_optional_string_with_shellexpand")] + pub index_store: Option, + + /// The average chunk size in bytes advertised to clients through the + /// `FastCDC` 2020 capability parameters and used for server-side + /// chunking in `SplitBlob`. Clients derive the minimum and maximum + /// chunk sizes from this value (avg / 4 and avg * 4). The value must + /// be between 1 KiB and 1 MiB. + /// + /// Default: 524288 (512 KiB) + #[serde(default)] + pub avg_chunk_size_bytes: u64, + + /// Maximum number of chunks accepted in a `SpliceBlob` request or + /// produced by on-demand chunking in `SplitBlob`. Blobs that would + /// produce more chunks are served without chunking (`SplitBlob` returns + /// `NOT_FOUND` and clients fall back to a regular download). This bounds + /// the size of stored chunk layouts and of `SplitBlobResponse` messages + /// (roughly 80-140 bytes per chunk). At the default average chunk size + /// the default cap supports blobs up to ~25 GiB; note that values above + /// ~50000 may produce responses that exceed default gRPC message size + /// limits on clients. + /// + /// Default: 50000 + #[serde(default)] + pub max_chunk_count: u64, +} + +impl CasChunkingConfig { + /// Default for `avg_chunk_size_bytes`, the value recommended by the + /// REAPI spec for `FastCdc2020Params`. + pub const DEFAULT_AVG_CHUNK_SIZE_BYTES: u64 = 512 * 1024; + /// Bounds for `avg_chunk_size_bytes` mandated by the REAPI spec for + /// `FastCdc2020Params`. + pub const MIN_AVG_CHUNK_SIZE_BYTES: u64 = 1024; + pub const MAX_AVG_CHUNK_SIZE_BYTES: u64 = 1024 * 1024; + /// Default for `max_chunk_count`. + pub const DEFAULT_MAX_CHUNK_COUNT: u64 = 50_000; + + /// Returns `avg_chunk_size_bytes` with the default applied. + #[must_use] + pub const fn resolved_avg_chunk_size_bytes(&self) -> u64 { + if self.avg_chunk_size_bytes == 0 { + Self::DEFAULT_AVG_CHUNK_SIZE_BYTES + } else { + self.avg_chunk_size_bytes + } + } + + /// Returns `max_chunk_count` with the default applied. + #[must_use] + pub const fn resolved_max_chunk_count(&self) -> u64 { + if self.max_chunk_count == 0 { + Self::DEFAULT_MAX_CHUNK_COUNT + } else { + self.max_chunk_count + } + } + + /// Returns `avg_chunk_size_bytes` with the default applied, or an error + /// when the configured value is outside the REAPI-mandated bounds. + pub fn validated_avg_chunk_size_bytes(&self) -> Result { + let avg_chunk_size_bytes = self.resolved_avg_chunk_size_bytes(); + if !(Self::MIN_AVG_CHUNK_SIZE_BYTES..=Self::MAX_AVG_CHUNK_SIZE_BYTES) + .contains(&avg_chunk_size_bytes) + { + return Err(make_err!( + Code::InvalidArgument, + "'experimental_chunking.avg_chunk_size_bytes' is {avg_chunk_size_bytes}, must be between {} and {}", + Self::MIN_AVG_CHUNK_SIZE_BYTES, + Self::MAX_AVG_CHUNK_SIZE_BYTES + )); + } + Ok(avg_chunk_size_bytes) + } } #[derive(Deserialize, Serialize, Debug, Default)] diff --git a/nativelink-proto/build/bazel/remote/execution/v2/remote_execution.proto b/nativelink-proto/build/bazel/remote/execution/v2/remote_execution.proto index ebfcea3c7..b23d01ae0 100644 --- a/nativelink-proto/build/bazel/remote/execution/v2/remote_execution.proto +++ b/nativelink-proto/build/bazel/remote/execution/v2/remote_execution.proto @@ -429,6 +429,125 @@ service ContentAddressableStorage { rpc GetTree(GetTreeRequest) returns (stream GetTreeResponse) { option (google.api.http) = { get: "/v2/{instance_name=**}/blobs/{root_digest.hash}/{root_digest.size_bytes}:getTree" }; } + + // SplitBlob retrieves information about how a blob is split into chunks. + // + // This call returns information about how a blob is split into chunks, and + // returns a list of the chunk digests. Using the returned list of chunk digests, + // a client can check which chunks are locally available and only fetch the + // missing ones. The desired blob can be assembled by concatenating the fetched + // chunks in the order of the digests in the list. The chunks SHOULD all be + // available in the CAS. + // + // This API can be used to reduce the required data to download a large blob + // from CAS if some chunks from similar blobs are locally available. For this + // procedure to work properly, blobs SHOULD be split in a content-defined way, + // rather than with fixed-sized chunking. + // + // If a split request is answered successfully, a client can expect the + // following guarantees from the server: + // 1. The blob chunks are stored in CAS. + // 2. Concatenating the blob chunks in the order of the digest list returned + // by the server results in the original blob. + // + // Servers which implement this functionality MUST declare that they support + // it by setting the + // [CacheCapabilities.split_blob_support][build.bazel.remote.execution.v2.CacheCapabilities.split_blob_support] + // field accordingly. + // + // Clients MUST check that the server supports this capability, before using + // it. + // + // Clients SHOULD verify that the digest of the blob assembled by the fetched + // chunks is equal to the requested blob digest. + // + // The lifetimes of the generated chunk blobs MAY be independent of the + // lifetime of the original blob. In particular: + // * A blob and any chunk derived from it MAY be evicted from the CAS at + // different times. + // * A call to [SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + // extends the lifetime of the original blob, and sets the lifetimes of + // the resulting chunks (or extends the lifetimes of already-existing + // chunks). + // * Touching a chunk extends its lifetime, but the server MAY choose not + // to extend the lifetime of the original blob. + // * Touching the original blob extends its lifetime, but the server MAY + // choose not to extend the lifetimes of chunks derived from it. + // + // When blob splitting and splicing is used at the same time, the clients and + // the server SHOULD agree out-of-band upon a chunking algorithm used by both + // parties to benefit from each other's chunk data and avoid unnecessary data + // duplication. + // + // Errors: + // + // * `NOT_FOUND`: The requested blob is not present in the CAS, OR there is no + // split information available for the blob, OR at least one chunk needed to + // reconstruct the blob is missing from the CAS. + // * `RESOURCE_EXHAUSTED`: There is insufficient disk quota to store the blob + // chunks. + rpc SplitBlob(SplitBlobRequest) returns (SplitBlobResponse) { + option (google.api.http) = { get: "/v2/{instance_name=**}/blobs/{blob_digest.hash}/{blob_digest.size_bytes}:splitBlob" }; + } + + // SpliceBlob tells the CAS how chunks can compose a blob. + // + // This is the complementary operation to the + // [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + // function to handle the chunked upload of large blobs to save upload + // traffic. + // + // When uploading a large blob using chunked upload, clients MUST first upload + // all chunks to the CAS, then call this RPC to tell the server how those chunks + // compose the original blob. The chunks referenced in the SpliceBlob call SHOULD be + // available in the CAS before calling this RPC. + // + // If a client needs to upload a large blob and is able to split a blob into + // chunks in such a way that reusable chunks are obtained, e.g., by means of + // content-defined chunking, it can first determine which parts of the blob + // are already available in the remote CAS and upload the missing chunks, and + // then use this API to store information on how the chunks compose the + // original blob. + // + // Servers which implement this functionality MUST declare that they support + // it by setting the + // [CacheCapabilities.splice_blob_support][build.bazel.remote.execution.v2.CacheCapabilities.splice_blob_support] + // field accordingly. + // + // Clients MUST check that the server supports this capability, before using + // it. + // + // In order to ensure data consistency of the CAS, the server MUST only add + // blobs to the CAS after verifying their digests. In particular, servers MUST NOT + // trust digests provided by the client. The server MAY accept a request as no-op + // if the client-specified blob is already in CAS or if information on how to + // construct the blob from chunks is available. If the client-specified blob is + // not already in the CAS, the server MUST verify that the digest of the newly + // created blob assembled from chunks matches the digest specified by the + // client, and reject the request if they differ. Servers MAY choose to allow + // overwriting existing chunk mappings or to store multiple chunk mappings for + // the same blob. + // + // When blob splitting and splicing is used at the same time, the clients and + // the server SHOULD agree out-of-band upon a chunking algorithm used by both + // parties to benefit from each other's chunk data and avoid unnecessary data + // duplication. + // + // Errors: + // + // * `NOT_FOUND`: At least one of the blob chunks is not present in the CAS. + // * `RESOURCE_EXHAUSTED`: There is insufficient disk quota to store the + // spliced blob. + // * `INVALID_ARGUMENT`: The digest of the spliced blob is different from the + // provided expected digest. + // * `ALREADY_EXISTS`: The blob already exists in CAS and the server did not + // extend the lifetime of the chunks specified in the request, e.g. because + // it prefers a different chunking and extended those instead. Clients can + // call [SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + // to check what chunk mapping the server is using. + rpc SpliceBlob(SpliceBlobRequest) returns (SpliceBlobResponse) { + option (google.api.http) = { post: "/v2/{instance_name=**}/blobs:spliceBlob" body: "*" }; + } } // The Capabilities service may be used by remote execution clients to query @@ -1777,6 +1896,100 @@ message GetTreeResponse { string next_page_token = 2; } +// A request message for +// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob]. +message SplitBlobRequest { + // The instance of the execution system to operate against. A server may + // support multiple instances of the execution system (with their own workers, + // storage, caches, etc.). The server MAY require use of this field to select + // between them in an implementation-defined fashion, otherwise it can be + // omitted. + string instance_name = 1; + + // The digest of the blob to be split. + Digest blob_digest = 2; + + // The digest function of the blob to be split. + // + // If the digest function used is one of MD5, MURMUR3, SHA1, SHA256, + // SHA384, SHA512, or VSO, the client MAY leave this field unset. In + // that case the server SHOULD infer the digest function using the + // length of the blob digest hashes and the digest functions announced + // in the server's capabilities. + DigestFunction.Value digest_function = 3; + + // The chunking function that the client prefers to use. + // + // The server MAY use a different chunking function. + ChunkingFunction.Value chunking_function = 4; +} + +// A response message for +// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob]. +message SplitBlobResponse { + // The ordered list of digests of the chunks into which the blob was split. + // The original blob is assembled by concatenating the chunk data according to + // the order of the digests given by this list. + // + // The server MUST use the same digest function as the one explicitly or + // implicitly (through hash length) specified in the split request. + repeated Digest chunk_digests = 1; + + // The chunking function used to split the blob. + ChunkingFunction.Value chunking_function = 2; +} + +// A request message for +// [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob]. +message SpliceBlobRequest { + // The instance of the execution system to operate against. A server may + // support multiple instances of the execution system (with their own workers, + // storage, caches, etc.). The server MAY require use of this field to select + // between them in an implementation-defined fashion, otherwise it can be + // omitted. + string instance_name = 1; + + // Expected digest of the spliced blob. The client MUST set this field due + // to the following reasons: + // 1. It allows the server to perform an early existence check of the blob + // or existing chunks that assemble the blob before spending the splicing + // effort, as described in the [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob] + // documentation. + // 2. It allows servers with different storage backends to dispatch the + // request to the correct storage backend based on the size and/or the + // hash of the blob. + // 3. If chunking information already exists for the blob, it allows + // the server to keep the existing chunking information or replace it with + // new chunking information. + Digest blob_digest = 2; + + // The ordered list of digests of the chunks which need to be concatenated to + // assemble the original blob. + repeated Digest chunk_digests = 3; + + // The digest function of all chunks to be concatenated and of the blob to be + // spliced. The server MUST use the same digest function for both cases. + // + // If the digest function used is one of MD5, MURMUR3, SHA1, SHA256, SHA384, + // SHA512, or VSO, the client MAY leave this field unset. In that case the + // server SHOULD infer the digest function using the length of the blob digest + // hashes and the digest functions announced in the server's capabilities. + DigestFunction.Value digest_function = 4; + + // The chunking function that the client used to split the blob. + ChunkingFunction.Value chunking_function = 5; +} + +// A response message for +// [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob]. +message SpliceBlobResponse { + // Computed digest of the spliced blob. + // + // The server MUST use the same digest function as the one explicitly or + // implicitly (through hash length) specified in the splice request. + Digest blob_digest = 1; +} + // A request message for // [Capabilities.GetCapabilities][build.bazel.remote.execution.v2.Capabilities.GetCapabilities]. message GetCapabilitiesRequest { @@ -1967,6 +2180,34 @@ message Compressor { } } +// The chunking function is used to split a blob into chunks. +// +// The server advertises support for a chunking function by setting the +// corresponding params field in +// [CacheCapabilities][build.bazel.remote.execution.v2.CacheCapabilities]. +// For example, if fast_cdc_2020_params is set, the server supports FAST_CDC_2020. +// +// For optimal deduplication, clients SHOULD use an advertised chunking function. +// When clients use UNKNOWN, the server chooses an algorithm for SplitBlob and +// simply verifies chunk concatenation for SpliceBlob. +message ChunkingFunction { + enum Value { + // No specific algorithm. Servers MUST always accept this value. + // For SplitBlob, the server chooses the algorithm. For SpliceBlob, the + // server only verifies that chunks concatenate to form the expected blob. + UNKNOWN = 0; + + // The FastCDC chunking algorithm as described in the 2020 paper by + // Wen Xia, et al. See https://ieeexplore.ieee.org/document/9055082 + // for details. + FAST_CDC_2020 = 1; + + // The RepMaxCDC chunking algorithm as implemented by buildbarn/go-cdc. + // See https://github.com/buildbarn/go-cdc for details. + REP_MAX_CDC = 2; + } +} + // Capabilities of the remote cache system. message CacheCapabilities { // All the digest functions supported by the remote cache. @@ -2000,6 +2241,121 @@ message CacheCapabilities { // [BatchUpdateBlobs][build.bazel.remote.execution.v2.ContentAddressableStorage.BatchUpdateBlobs] // requests. repeated Compressor.Value supported_batch_update_compressors = 7; + + // The maximum blob size that the server will accept for CAS blob uploads. + // - If it is 0, it means there is no limit set. A client may assume + // arbitrarily large blobs may be uploaded to and downloaded from the cache. + // - If it is larger than 0, implementations SHOULD NOT attempt to upload + // blobs with size larger than the limit. Servers SHOULD reject blob + // uploads over the `max_cas_blob_size_bytes` limit with response code + // `INVALID_ARGUMENT` + // - If the cache implementation returns a given limit, it MAY still serve + // blobs larger than this limit. + int64 max_cas_blob_size_bytes = 8; + + // Whether blob splitting is supported for the particular server/instance. If + // yes, the server/instance implements the specified behavior for blob + // splitting and a meaningful result can be expected from the + // [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + // operation. + bool split_blob_support = 9; + + // Whether blob splicing is supported for the particular server/instance. If + // yes, the server/instance implements the specified behavior for blob + // splicing and a meaningful result can be expected from the + // [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob] + // operation. + bool splice_blob_support = 10; + + // The parameters for the FastCDC 2020 chunking algorithm. + // If set, the server supports the FastCDC chunking algorithm. + FastCdc2020Params fast_cdc_2020_params = 11; + + // The parameters for the RepMaxCDC chunking algorithm. + // If set, the server supports the RepMaxCDC chunking algorithm. + RepMaxCdcParams rep_max_cdc_params = 12; +} + +// Parameters for the FastCDC content-defined chunking algorithm. +// +// Implementations MUST follow the FastCDC 2020 paper by Wen Xia, et al.: +// https://ieeexplore.ieee.org/document/9055082 +// +// Supported implementations: +// - Rust: https://docs.rs/fastcdc/3.2.1/fastcdc/v2020/index.html +// - Go: https://github.com/buildbuddy-io/fastcdc2020 +// +// Test vectors can be found in the accompanying fastcdc2020_test_vectors.txt file. +// +// Implementations MUST use normalization level 2, which has been found +// successful for build artifacts with an average chunk size of 512 KiB. +// +// Key algorithm components from the paper: +// +// GEAR table: 256 64-bit integers for the rolling hash, computed as: +// GEAR[i] = high_64_bits(MD5(byte(i))) for i in 0..255 +// +// MASKS table: Bit patterns for chunk boundary detection, derived from +// the C reference implementation. The mask selection based on average +// chunk size SHOULD match the paper. +// +// The minimum and maximum chunk sizes MUST be derived from the average: +// - min_chunk_size = avg_chunk_size_bytes / 4 +// - max_chunk_size = avg_chunk_size_bytes * 4 +// +// Blobs smaller than max_chunk_size (avg_chunk_size_bytes * 4) SHOULD be +// uploaded without chunking. +// +// If any of the advertised parameters are not within the expected range, +// the client SHOULD ignore FastCDC chunking function support. +message FastCdc2020Params { + // The average (expected) chunk size for the FastCDC chunking algorithm. + // The value MUST be between 1 KiB and 1 MiB. The recommended value is + // 524288 (512 KiB). + uint64 avg_chunk_size_bytes = 1; + + // The seed for the FastCDC mask generation. + // The recommended value is 0. + // + // All clients sharing a cache SHOULD use the same seed to maximize + // chunk reuse. + uint32 seed = 2; +} + +// Parameters for the RepMaxCDC content-defined chunking algorithm. +// +// Supported implementations: +// - Go: https://github.com/buildbarn/go-cdc +// +// Key algorithm components: +// +// GEAR table: 256 64-bit integers for the rolling hash, computed as: +// GEAR[i] = high_64_bits(MD5(byte(i))) for i in 0..255 +// +// The algorithm repeatedly applies chunking until all chunks are in the +// range [min_chunk_size_bytes, 2*min_chunk_size_bytes). Cutting points are +// selected where the Gear rolling hash is maximized within a lookahead +// window of horizon_size_bytes. +// +// For sufficiently large files, the average chunk size prior to +// deduplication will approximately be min_chunk_size_bytes divided by +// Rényi's parking constant (0.7475979203...). More details: +// https://mathworld.wolfram.com/RenyisParkingConstants.html +// +// If any of the advertised parameters are not within the expected range, +// the client SHOULD ignore RepMaxCDC chunking function support. +message RepMaxCdcParams { + // The minimum chunk size for the RepMaxCDC chunking algorithm. + // The value MUST be at least 64 bytes (the Gear hash window size). + // All chunks will be in the range [min_chunk_size_bytes, 2*min_chunk_size_bytes). + // The recommended value is 262144 (256 KiB). + uint64 min_chunk_size_bytes = 1; + + // The lookahead window for finding optimal cutting points. + // Larger values improve deduplication quality with diminishing returns. + // Setting to 0 produces uniform chunks of min_chunk_size_bytes. + // The recommended value is 8 * min_chunk_size_bytes. + uint64 horizon_size_bytes = 2; } // Capabilities of the remote execution system. diff --git a/nativelink-proto/genproto/build.bazel.remote.execution.v2.pb.rs b/nativelink-proto/genproto/build.bazel.remote.execution.v2.pb.rs index b7c8f6634..f4b7c43e7 100644 --- a/nativelink-proto/genproto/build.bazel.remote.execution.v2.pb.rs +++ b/nativelink-proto/genproto/build.bazel.remote.execution.v2.pb.rs @@ -1438,6 +1438,104 @@ pub struct GetTreeResponse { pub next_page_token: ::prost::alloc::string::String, } /// A request message for +/// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob]. +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SplitBlobRequest { + /// The instance of the execution system to operate against. A server may + /// support multiple instances of the execution system (with their own workers, + /// storage, caches, etc.). The server MAY require use of this field to select + /// between them in an implementation-defined fashion, otherwise it can be + /// omitted. + #[prost(string, tag = "1")] + pub instance_name: ::prost::alloc::string::String, + /// The digest of the blob to be split. + #[prost(message, optional, tag = "2")] + pub blob_digest: ::core::option::Option, + /// The digest function of the blob to be split. + /// + /// If the digest function used is one of MD5, MURMUR3, SHA1, SHA256, + /// SHA384, SHA512, or VSO, the client MAY leave this field unset. In + /// that case the server SHOULD infer the digest function using the + /// length of the blob digest hashes and the digest functions announced + /// in the server's capabilities. + #[prost(enumeration = "digest_function::Value", tag = "3")] + pub digest_function: i32, + /// The chunking function that the client prefers to use. + /// + /// The server MAY use a different chunking function. + #[prost(enumeration = "chunking_function::Value", tag = "4")] + pub chunking_function: i32, +} +/// A response message for +/// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob]. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SplitBlobResponse { + /// The ordered list of digests of the chunks into which the blob was split. + /// The original blob is assembled by concatenating the chunk data according to + /// the order of the digests given by this list. + /// + /// The server MUST use the same digest function as the one explicitly or + /// implicitly (through hash length) specified in the split request. + #[prost(message, repeated, tag = "1")] + pub chunk_digests: ::prost::alloc::vec::Vec, + /// The chunking function used to split the blob. + #[prost(enumeration = "chunking_function::Value", tag = "2")] + pub chunking_function: i32, +} +/// A request message for +/// [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob]. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SpliceBlobRequest { + /// The instance of the execution system to operate against. A server may + /// support multiple instances of the execution system (with their own workers, + /// storage, caches, etc.). The server MAY require use of this field to select + /// between them in an implementation-defined fashion, otherwise it can be + /// omitted. + #[prost(string, tag = "1")] + pub instance_name: ::prost::alloc::string::String, + /// Expected digest of the spliced blob. The client MUST set this field due + /// to the following reasons: + /// 1. It allows the server to perform an early existence check of the blob + /// or existing chunks that assemble the blob before spending the splicing + /// effort, as described in the [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob] + /// documentation. + /// 2. It allows servers with different storage backends to dispatch the + /// request to the correct storage backend based on the size and/or the + /// hash of the blob. + /// 3. If chunking information already exists for the blob, it allows + /// the server to keep the existing chunking information or replace it with + /// new chunking information. + #[prost(message, optional, tag = "2")] + pub blob_digest: ::core::option::Option, + /// The ordered list of digests of the chunks which need to be concatenated to + /// assemble the original blob. + #[prost(message, repeated, tag = "3")] + pub chunk_digests: ::prost::alloc::vec::Vec, + /// The digest function of all chunks to be concatenated and of the blob to be + /// spliced. The server MUST use the same digest function for both cases. + /// + /// If the digest function used is one of MD5, MURMUR3, SHA1, SHA256, SHA384, + /// SHA512, or VSO, the client MAY leave this field unset. In that case the + /// server SHOULD infer the digest function using the length of the blob digest + /// hashes and the digest functions announced in the server's capabilities. + #[prost(enumeration = "digest_function::Value", tag = "4")] + pub digest_function: i32, + /// The chunking function that the client used to split the blob. + #[prost(enumeration = "chunking_function::Value", tag = "5")] + pub chunking_function: i32, +} +/// A response message for +/// [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob]. +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SpliceBlobResponse { + /// Computed digest of the spliced blob. + /// + /// The server MUST use the same digest function as the one explicitly or + /// implicitly (through hash length) specified in the splice request. + #[prost(message, optional, tag = "1")] + pub blob_digest: ::core::option::Option, +} +/// A request message for /// [Capabilities.GetCapabilities][build.bazel.remote.execution.v2.Capabilities.GetCapabilities]. #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct GetCapabilitiesRequest { @@ -1749,6 +1847,68 @@ pub mod compressor { } } } +/// The chunking function is used to split a blob into chunks. +/// +/// The server advertises support for a chunking function by setting the +/// corresponding params field in +/// [CacheCapabilities][build.bazel.remote.execution.v2.CacheCapabilities]. +/// For example, if fast_cdc_2020_params is set, the server supports FAST_CDC_2020. +/// +/// For optimal deduplication, clients SHOULD use an advertised chunking function. +/// When clients use UNKNOWN, the server chooses an algorithm for SplitBlob and +/// simply verifies chunk concatenation for SpliceBlob. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct ChunkingFunction {} +/// Nested message and enum types in `ChunkingFunction`. +pub mod chunking_function { + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] + #[repr(i32)] + pub enum Value { + /// No specific algorithm. Servers MUST always accept this value. + /// For SplitBlob, the server chooses the algorithm. For SpliceBlob, the + /// server only verifies that chunks concatenate to form the expected blob. + Unknown = 0, + /// The FastCDC chunking algorithm as described in the 2020 paper by + /// Wen Xia, et al. See + /// for details. + FastCdc2020 = 1, + /// The RepMaxCDC chunking algorithm as implemented by buildbarn/go-cdc. + /// See for details. + RepMaxCdc = 2, + } + impl Value { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Unknown => "UNKNOWN", + Self::FastCdc2020 => "FAST_CDC_2020", + Self::RepMaxCdc => "REP_MAX_CDC", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "UNKNOWN" => Some(Self::Unknown), + "FAST_CDC_2020" => Some(Self::FastCdc2020), + "REP_MAX_CDC" => Some(Self::RepMaxCdc), + _ => None, + } + } + } +} /// Capabilities of the remote cache system. #[derive(Clone, PartialEq, ::prost::Message)] pub struct CacheCapabilities { @@ -1786,6 +1946,123 @@ pub struct CacheCapabilities { /// requests. #[prost(enumeration = "compressor::Value", repeated, tag = "7")] pub supported_batch_update_compressors: ::prost::alloc::vec::Vec, + /// The maximum blob size that the server will accept for CAS blob uploads. + /// - If it is 0, it means there is no limit set. A client may assume + /// arbitrarily large blobs may be uploaded to and downloaded from the cache. + /// - If it is larger than 0, implementations SHOULD NOT attempt to upload + /// blobs with size larger than the limit. Servers SHOULD reject blob + /// uploads over the `max_cas_blob_size_bytes` limit with response code + /// `INVALID_ARGUMENT` + /// - If the cache implementation returns a given limit, it MAY still serve + /// blobs larger than this limit. + #[prost(int64, tag = "8")] + pub max_cas_blob_size_bytes: i64, + /// Whether blob splitting is supported for the particular server/instance. If + /// yes, the server/instance implements the specified behavior for blob + /// splitting and a meaningful result can be expected from the + /// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// operation. + #[prost(bool, tag = "9")] + pub split_blob_support: bool, + /// Whether blob splicing is supported for the particular server/instance. If + /// yes, the server/instance implements the specified behavior for blob + /// splicing and a meaningful result can be expected from the + /// [ContentAddressableStorage.SpliceBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SpliceBlob] + /// operation. + #[prost(bool, tag = "10")] + pub splice_blob_support: bool, + /// The parameters for the FastCDC 2020 chunking algorithm. + /// If set, the server supports the FastCDC chunking algorithm. + #[prost(message, optional, tag = "11")] + pub fast_cdc_2020_params: ::core::option::Option, + /// The parameters for the RepMaxCDC chunking algorithm. + /// If set, the server supports the RepMaxCDC chunking algorithm. + #[prost(message, optional, tag = "12")] + pub rep_max_cdc_params: ::core::option::Option, +} +/// Parameters for the FastCDC content-defined chunking algorithm. +/// +/// Implementations MUST follow the FastCDC 2020 paper by Wen Xia, et al.: +/// +/// +/// Supported implementations: +/// - Rust: +/// - Go: +/// +/// Test vectors can be found in the accompanying fastcdc2020_test_vectors.txt file. +/// +/// Implementations MUST use normalization level 2, which has been found +/// successful for build artifacts with an average chunk size of 512 KiB. +/// +/// Key algorithm components from the paper: +/// +/// GEAR table: 256 64-bit integers for the rolling hash, computed as: +/// GEAR\[i\] = high_64_bits(MD5(byte(i))) for i in 0..255 +/// +/// MASKS table: Bit patterns for chunk boundary detection, derived from +/// the C reference implementation. The mask selection based on average +/// chunk size SHOULD match the paper. +/// +/// The minimum and maximum chunk sizes MUST be derived from the average: +/// - min_chunk_size = avg_chunk_size_bytes / 4 +/// - max_chunk_size = avg_chunk_size_bytes * 4 +/// +/// Blobs smaller than max_chunk_size (avg_chunk_size_bytes * 4) SHOULD be +/// uploaded without chunking. +/// +/// If any of the advertised parameters are not within the expected range, +/// the client SHOULD ignore FastCDC chunking function support. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct FastCdc2020Params { + /// The average (expected) chunk size for the FastCDC chunking algorithm. + /// The value MUST be between 1 KiB and 1 MiB. The recommended value is + /// 524288 (512 KiB). + #[prost(uint64, tag = "1")] + pub avg_chunk_size_bytes: u64, + /// The seed for the FastCDC mask generation. + /// The recommended value is 0. + /// + /// All clients sharing a cache SHOULD use the same seed to maximize + /// chunk reuse. + #[prost(uint32, tag = "2")] + pub seed: u32, +} +/// Parameters for the RepMaxCDC content-defined chunking algorithm. +/// +/// Supported implementations: +/// - Go: +/// +/// Key algorithm components: +/// +/// GEAR table: 256 64-bit integers for the rolling hash, computed as: +/// GEAR\[i\] = high_64_bits(MD5(byte(i))) for i in 0..255 +/// +/// The algorithm repeatedly applies chunking until all chunks are in the +/// range [min_chunk_size_bytes, 2*min_chunk_size_bytes). Cutting points are +/// selected where the Gear rolling hash is maximized within a lookahead +/// window of horizon_size_bytes. +/// +/// For sufficiently large files, the average chunk size prior to +/// deduplication will approximately be min_chunk_size_bytes divided by +/// Rényi's parking constant (0.7475979203...). More details: +/// +/// +/// If any of the advertised parameters are not within the expected range, +/// the client SHOULD ignore RepMaxCDC chunking function support. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct RepMaxCdcParams { + /// The minimum chunk size for the RepMaxCDC chunking algorithm. + /// The value MUST be at least 64 bytes (the Gear hash window size). + /// All chunks will be in the range [min_chunk_size_bytes, 2*min_chunk_size_bytes). + /// The recommended value is 262144 (256 KiB). + #[prost(uint64, tag = "1")] + pub min_chunk_size_bytes: u64, + /// The lookahead window for finding optimal cutting points. + /// Larger values improve deduplication quality with diminishing returns. + /// Setting to 0 produces uniform chunks of min_chunk_size_bytes. + /// The recommended value is 8 * min_chunk_size_bytes. + #[prost(uint64, tag = "2")] + pub horizon_size_bytes: u64, } /// Capabilities of the remote execution system. #[derive(Clone, PartialEq, ::prost::Message)] @@ -3340,6 +3617,175 @@ pub mod content_addressable_storage_client { ); self.inner.server_streaming(req, path, codec).await } + /// SplitBlob retrieves information about how a blob is split into chunks. + /// + /// This call returns information about how a blob is split into chunks, and + /// returns a list of the chunk digests. Using the returned list of chunk digests, + /// a client can check which chunks are locally available and only fetch the + /// missing ones. The desired blob can be assembled by concatenating the fetched + /// chunks in the order of the digests in the list. The chunks SHOULD all be + /// available in the CAS. + /// + /// This API can be used to reduce the required data to download a large blob + /// from CAS if some chunks from similar blobs are locally available. For this + /// procedure to work properly, blobs SHOULD be split in a content-defined way, + /// rather than with fixed-sized chunking. + /// + /// If a split request is answered successfully, a client can expect the + /// following guarantees from the server: + /// 1. The blob chunks are stored in CAS. + /// 2. Concatenating the blob chunks in the order of the digest list returned + /// by the server results in the original blob. + /// + /// Servers which implement this functionality MUST declare that they support + /// it by setting the + /// [CacheCapabilities.split_blob_support][build.bazel.remote.execution.v2.CacheCapabilities.split_blob_support] + /// field accordingly. + /// + /// Clients MUST check that the server supports this capability, before using + /// it. + /// + /// Clients SHOULD verify that the digest of the blob assembled by the fetched + /// chunks is equal to the requested blob digest. + /// + /// The lifetimes of the generated chunk blobs MAY be independent of the + /// lifetime of the original blob. In particular: + /// * A blob and any chunk derived from it MAY be evicted from the CAS at + /// different times. + /// * A call to [SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// extends the lifetime of the original blob, and sets the lifetimes of + /// the resulting chunks (or extends the lifetimes of already-existing + /// chunks). + /// * Touching a chunk extends its lifetime, but the server MAY choose not + /// to extend the lifetime of the original blob. + /// * Touching the original blob extends its lifetime, but the server MAY + /// choose not to extend the lifetimes of chunks derived from it. + /// + /// When blob splitting and splicing is used at the same time, the clients and + /// the server SHOULD agree out-of-band upon a chunking algorithm used by both + /// parties to benefit from each other's chunk data and avoid unnecessary data + /// duplication. + /// + /// Errors: + /// + /// * `NOT_FOUND`: The requested blob is not present in the CAS, OR there is no + /// split information available for the blob, OR at least one chunk needed to + /// reconstruct the blob is missing from the CAS. + /// * `RESOURCE_EXHAUSTED`: There is insufficient disk quota to store the blob + /// chunks. + pub async fn split_blob( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/build.bazel.remote.execution.v2.ContentAddressableStorage/SplitBlob", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "build.bazel.remote.execution.v2.ContentAddressableStorage", + "SplitBlob", + ), + ); + self.inner.unary(req, path, codec).await + } + /// SpliceBlob tells the CAS how chunks can compose a blob. + /// + /// This is the complementary operation to the + /// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// function to handle the chunked upload of large blobs to save upload + /// traffic. + /// + /// When uploading a large blob using chunked upload, clients MUST first upload + /// all chunks to the CAS, then call this RPC to tell the server how those chunks + /// compose the original blob. The chunks referenced in the SpliceBlob call SHOULD be + /// available in the CAS before calling this RPC. + /// + /// If a client needs to upload a large blob and is able to split a blob into + /// chunks in such a way that reusable chunks are obtained, e.g., by means of + /// content-defined chunking, it can first determine which parts of the blob + /// are already available in the remote CAS and upload the missing chunks, and + /// then use this API to store information on how the chunks compose the + /// original blob. + /// + /// Servers which implement this functionality MUST declare that they support + /// it by setting the + /// [CacheCapabilities.splice_blob_support][build.bazel.remote.execution.v2.CacheCapabilities.splice_blob_support] + /// field accordingly. + /// + /// Clients MUST check that the server supports this capability, before using + /// it. + /// + /// In order to ensure data consistency of the CAS, the server MUST only add + /// blobs to the CAS after verifying their digests. In particular, servers MUST NOT + /// trust digests provided by the client. The server MAY accept a request as no-op + /// if the client-specified blob is already in CAS or if information on how to + /// construct the blob from chunks is available. If the client-specified blob is + /// not already in the CAS, the server MUST verify that the digest of the newly + /// created blob assembled from chunks matches the digest specified by the + /// client, and reject the request if they differ. Servers MAY choose to allow + /// overwriting existing chunk mappings or to store multiple chunk mappings for + /// the same blob. + /// + /// When blob splitting and splicing is used at the same time, the clients and + /// the server SHOULD agree out-of-band upon a chunking algorithm used by both + /// parties to benefit from each other's chunk data and avoid unnecessary data + /// duplication. + /// + /// Errors: + /// + /// * `NOT_FOUND`: At least one of the blob chunks is not present in the CAS. + /// * `RESOURCE_EXHAUSTED`: There is insufficient disk quota to store the + /// spliced blob. + /// * `INVALID_ARGUMENT`: The digest of the spliced blob is different from the + /// provided expected digest. + /// * `ALREADY_EXISTS`: The blob already exists in CAS and the server did not + /// extend the lifetime of the chunks specified in the request, e.g. because + /// it prefers a different chunking and extended those instead. Clients can + /// call [SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// to check what chunk mapping the server is using. + pub async fn splice_blob( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/build.bazel.remote.execution.v2.ContentAddressableStorage/SpliceBlob", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "build.bazel.remote.execution.v2.ContentAddressableStorage", + "SpliceBlob", + ), + ); + self.inner.unary(req, path, codec).await + } } } /// Generated server implementations. @@ -3461,6 +3907,131 @@ pub mod content_addressable_storage_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + /// SplitBlob retrieves information about how a blob is split into chunks. + /// + /// This call returns information about how a blob is split into chunks, and + /// returns a list of the chunk digests. Using the returned list of chunk digests, + /// a client can check which chunks are locally available and only fetch the + /// missing ones. The desired blob can be assembled by concatenating the fetched + /// chunks in the order of the digests in the list. The chunks SHOULD all be + /// available in the CAS. + /// + /// This API can be used to reduce the required data to download a large blob + /// from CAS if some chunks from similar blobs are locally available. For this + /// procedure to work properly, blobs SHOULD be split in a content-defined way, + /// rather than with fixed-sized chunking. + /// + /// If a split request is answered successfully, a client can expect the + /// following guarantees from the server: + /// 1. The blob chunks are stored in CAS. + /// 2. Concatenating the blob chunks in the order of the digest list returned + /// by the server results in the original blob. + /// + /// Servers which implement this functionality MUST declare that they support + /// it by setting the + /// [CacheCapabilities.split_blob_support][build.bazel.remote.execution.v2.CacheCapabilities.split_blob_support] + /// field accordingly. + /// + /// Clients MUST check that the server supports this capability, before using + /// it. + /// + /// Clients SHOULD verify that the digest of the blob assembled by the fetched + /// chunks is equal to the requested blob digest. + /// + /// The lifetimes of the generated chunk blobs MAY be independent of the + /// lifetime of the original blob. In particular: + /// * A blob and any chunk derived from it MAY be evicted from the CAS at + /// different times. + /// * A call to [SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// extends the lifetime of the original blob, and sets the lifetimes of + /// the resulting chunks (or extends the lifetimes of already-existing + /// chunks). + /// * Touching a chunk extends its lifetime, but the server MAY choose not + /// to extend the lifetime of the original blob. + /// * Touching the original blob extends its lifetime, but the server MAY + /// choose not to extend the lifetimes of chunks derived from it. + /// + /// When blob splitting and splicing is used at the same time, the clients and + /// the server SHOULD agree out-of-band upon a chunking algorithm used by both + /// parties to benefit from each other's chunk data and avoid unnecessary data + /// duplication. + /// + /// Errors: + /// + /// * `NOT_FOUND`: The requested blob is not present in the CAS, OR there is no + /// split information available for the blob, OR at least one chunk needed to + /// reconstruct the blob is missing from the CAS. + /// * `RESOURCE_EXHAUSTED`: There is insufficient disk quota to store the blob + /// chunks. + async fn split_blob( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// SpliceBlob tells the CAS how chunks can compose a blob. + /// + /// This is the complementary operation to the + /// [ContentAddressableStorage.SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// function to handle the chunked upload of large blobs to save upload + /// traffic. + /// + /// When uploading a large blob using chunked upload, clients MUST first upload + /// all chunks to the CAS, then call this RPC to tell the server how those chunks + /// compose the original blob. The chunks referenced in the SpliceBlob call SHOULD be + /// available in the CAS before calling this RPC. + /// + /// If a client needs to upload a large blob and is able to split a blob into + /// chunks in such a way that reusable chunks are obtained, e.g., by means of + /// content-defined chunking, it can first determine which parts of the blob + /// are already available in the remote CAS and upload the missing chunks, and + /// then use this API to store information on how the chunks compose the + /// original blob. + /// + /// Servers which implement this functionality MUST declare that they support + /// it by setting the + /// [CacheCapabilities.splice_blob_support][build.bazel.remote.execution.v2.CacheCapabilities.splice_blob_support] + /// field accordingly. + /// + /// Clients MUST check that the server supports this capability, before using + /// it. + /// + /// In order to ensure data consistency of the CAS, the server MUST only add + /// blobs to the CAS after verifying their digests. In particular, servers MUST NOT + /// trust digests provided by the client. The server MAY accept a request as no-op + /// if the client-specified blob is already in CAS or if information on how to + /// construct the blob from chunks is available. If the client-specified blob is + /// not already in the CAS, the server MUST verify that the digest of the newly + /// created blob assembled from chunks matches the digest specified by the + /// client, and reject the request if they differ. Servers MAY choose to allow + /// overwriting existing chunk mappings or to store multiple chunk mappings for + /// the same blob. + /// + /// When blob splitting and splicing is used at the same time, the clients and + /// the server SHOULD agree out-of-band upon a chunking algorithm used by both + /// parties to benefit from each other's chunk data and avoid unnecessary data + /// duplication. + /// + /// Errors: + /// + /// * `NOT_FOUND`: At least one of the blob chunks is not present in the CAS. + /// * `RESOURCE_EXHAUSTED`: There is insufficient disk quota to store the + /// spliced blob. + /// * `INVALID_ARGUMENT`: The digest of the spliced blob is different from the + /// provided expected digest. + /// * `ALREADY_EXISTS`: The blob already exists in CAS and the server did not + /// extend the lifetime of the chunks specified in the request, e.g. because + /// it prefers a different chunking and extended those instead. Clients can + /// call [SplitBlob][build.bazel.remote.execution.v2.ContentAddressableStorage.SplitBlob] + /// to check what chunk mapping the server is using. + async fn splice_blob( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } /// The CAS (content-addressable storage) is used to store the inputs to and /// outputs from the execution service. Each piece of content is addressed by the @@ -3876,6 +4447,104 @@ pub mod content_addressable_storage_server { }; Box::pin(fut) } + "/build.bazel.remote.execution.v2.ContentAddressableStorage/SplitBlob" => { + #[allow(non_camel_case_types)] + struct SplitBlobSvc(pub Arc); + impl< + T: ContentAddressableStorage, + > tonic::server::UnaryService + for SplitBlobSvc { + type Response = super::SplitBlobResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::split_blob( + &inner, + request, + ) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = SplitBlobSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/build.bazel.remote.execution.v2.ContentAddressableStorage/SpliceBlob" => { + #[allow(non_camel_case_types)] + struct SpliceBlobSvc(pub Arc); + impl< + T: ContentAddressableStorage, + > tonic::server::UnaryService + for SpliceBlobSvc { + type Response = super::SpliceBlobResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::splice_blob( + &inner, + request, + ) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = SpliceBlobSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } _ => { Box::pin(async move { let mut response = http::Response::new( diff --git a/nativelink-service/BUILD.bazel b/nativelink-service/BUILD.bazel index 2d13c305b..80bc06dd4 100644 --- a/nativelink-service/BUILD.bazel +++ b/nativelink-service/BUILD.bazel @@ -34,6 +34,7 @@ rust_library( "//nativelink-util", "@crates//:axum", "@crates//:bytes", + "@crates//:fastcdc", "@crates//:futures", "@crates//:http-body-util", "@crates//:hyper-1.7.0", @@ -46,6 +47,7 @@ rust_library( "@crates//:serde_json5", "@crates//:sha2", "@crates//:tokio", + "@crates//:tokio-util", "@crates//:tonic", "@crates//:tower", "@crates//:tracing", @@ -62,11 +64,15 @@ rust_test_suite( "tests/bytestream_server_test.rs", "tests/cas_server_test.rs", "tests/execution_server_test.rs", + "tests/fastcdc_conformance_test.rs", "tests/fetch_server_test.rs", "tests/health_server_test.rs", "tests/push_server_test.rs", "tests/worker_api_server_test.rs", ], + compile_data = glob(["tests/data/**"]) + [ + "//nativelink-util:tests/data/SekienAkashita.jpg", + ], proc_macro_deps = [ "//nativelink-macro", "@crates//:async-trait", @@ -83,6 +89,7 @@ rust_test_suite( "@crates//:async-lock", "@crates//:axum", "@crates//:bytes", + "@crates//:fastcdc", "@crates//:futures", "@crates//:hex", "@crates//:http-body-util", diff --git a/nativelink-service/Cargo.toml b/nativelink-service/Cargo.toml index e7369f8f8..13efb8a12 100644 --- a/nativelink-service/Cargo.toml +++ b/nativelink-service/Cargo.toml @@ -17,6 +17,7 @@ nativelink-util = { path = "../nativelink-util" } axum = { version = "0.8.3", default-features = false } bytes = { version = "1.10.1", default-features = false } +fastcdc = { version = "3.2.1", default-features = false, features = ["tokio"] } futures = { version = "0.3.31", default-features = false } http-body-util = { version = "0.1.3", default-features = false } hyper = { version = "1.6.0", default-features = false } @@ -43,6 +44,7 @@ tokio = { version = "1.52.2", features = [ tokio-stream = { version = "0.1.17", features = [ "fs", ], default-features = false } +tokio-util = { version = "0.7.14", features = ["io"], default-features = false } tonic = { version = "0.14.0", features = [ "gzip", "router", diff --git a/nativelink-service/src/capabilities_server.rs b/nativelink-service/src/capabilities_server.rs index 6e4102033..f2b32bfd4 100644 --- a/nativelink-service/src/capabilities_server.rs +++ b/nativelink-service/src/capabilities_server.rs @@ -15,7 +15,9 @@ use std::collections::HashMap; use std::sync::Arc; -use nativelink_config::cas_server::{CapabilitiesConfig, InstanceName, WithInstanceName}; +use nativelink_config::cas_server::{ + CapabilitiesConfig, CasStoreConfig, InstanceName, WithInstanceName, +}; use nativelink_error::{Error, ResultExt}; use nativelink_proto::build::bazel::remote::execution::v2::capabilities_server::{ Capabilities, CapabilitiesServer as Server, @@ -24,7 +26,7 @@ use nativelink_proto::build::bazel::remote::execution::v2::digest_function::Valu use nativelink_proto::build::bazel::remote::execution::v2::priority_capabilities::PriorityRange; use nativelink_proto::build::bazel::remote::execution::v2::symlink_absolute_path_strategy::Value as SymlinkAbsolutePathStrategy; use nativelink_proto::build::bazel::remote::execution::v2::{ - ActionCacheUpdateCapabilities, CacheCapabilities, ExecutionCapabilities, + ActionCacheUpdateCapabilities, CacheCapabilities, ExecutionCapabilities, FastCdc2020Params, GetCapabilitiesRequest, PriorityCapabilities, ServerCapabilities, }; use nativelink_proto::build::bazel::semver::SemVer; @@ -38,13 +40,36 @@ const MAX_BATCH_TOTAL_SIZE: i64 = 64 * 1024; #[derive(Debug, Default)] pub struct CapabilitiesServer { supported_node_properties_for_instance: HashMap>, + chunking_params_for_instance: HashMap, } impl CapabilitiesServer { pub async fn new( configs: &[WithInstanceName], scheduler_map: &HashMap>, + cas_configs: &[WithInstanceName], ) -> Result { + let mut chunking_params_for_instance = HashMap::new(); + for cas_config in cas_configs { + if let Some(chunking_config) = &cas_config.experimental_chunking { + let avg_chunk_size_bytes = chunking_config + .validated_avg_chunk_size_bytes() + .err_tip(|| { + format!( + "In 'experimental_chunking' of instance '{}'", + cas_config.instance_name + ) + })?; + chunking_params_for_instance.insert( + cas_config.instance_name.clone(), + FastCdc2020Params { + avg_chunk_size_bytes, + seed: 0, + }, + ); + } + } + let mut supported_node_properties_for_instance = HashMap::new(); for config in configs { let mut properties = Vec::new(); @@ -75,6 +100,7 @@ impl CapabilitiesServer { } Ok(Self { supported_node_properties_for_instance, + chunking_params_for_instance, }) } @@ -119,6 +145,7 @@ impl Capabilities for CapabilitiesServer { ], }); + let chunking_params = self.chunking_params_for_instance.get(&instance_name); let resp = ServerCapabilities { cache_capabilities: Some(CacheCapabilities { digest_functions: vec![ @@ -133,6 +160,11 @@ impl Capabilities for CapabilitiesServer { symlink_absolute_path_strategy: SymlinkAbsolutePathStrategy::Disallowed.into(), supported_compressors: vec![], supported_batch_update_compressors: vec![], + max_cas_blob_size_bytes: 0, + split_blob_support: chunking_params.is_some(), + splice_blob_support: chunking_params.is_some(), + fast_cdc_2020_params: chunking_params.copied(), + rep_max_cdc_params: None, }), execution_capabilities, deprecated_api_version: None, diff --git a/nativelink-service/src/cas_server.rs b/nativelink-service/src/cas_server.rs index 68e146686..f79df7f12 100644 --- a/nativelink-service/src/cas_server.rs +++ b/nativelink-service/src/cas_server.rs @@ -13,38 +13,165 @@ // limitations under the License. use core::convert::Into; -use core::pin::Pin; +use core::pin::{Pin, pin}; +use core::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use core::time::Duration; use std::collections::{HashMap, VecDeque}; use bytes::Bytes; +use fastcdc::v2020::{AsyncStreamCDC, Normalization}; use futures::stream::{FuturesUnordered, Stream}; use futures::{StreamExt, TryStreamExt}; use nativelink_config::cas_server::{CasStoreConfig, WithInstanceName}; use nativelink_error::{Code, Error, ResultExt, error_if, make_err, make_input_err}; +use nativelink_metric::{ + MetricFieldData, MetricKind, MetricPublishKnownKindData, MetricsComponent, group, publish, +}; use nativelink_proto::build::bazel::remote::execution::v2::content_addressable_storage_server::{ ContentAddressableStorage, ContentAddressableStorageServer as Server, }; use nativelink_proto::build::bazel::remote::execution::v2::{ BatchReadBlobsRequest, BatchReadBlobsResponse, BatchUpdateBlobsRequest, - BatchUpdateBlobsResponse, Directory, FindMissingBlobsRequest, FindMissingBlobsResponse, - GetTreeRequest, GetTreeResponse, batch_read_blobs_response, batch_update_blobs_response, - compressor, + BatchUpdateBlobsResponse, Digest, Directory, FindMissingBlobsRequest, FindMissingBlobsResponse, + GetTreeRequest, GetTreeResponse, SpliceBlobRequest, SpliceBlobResponse, SplitBlobRequest, + SplitBlobResponse, batch_read_blobs_response, batch_update_blobs_response, chunking_function, + compressor, digest_function, }; use nativelink_proto::google::rpc::Status as GrpcStatus; use nativelink_store::ac_utils::get_and_decode_digest; use nativelink_store::grpc_store::GrpcStore; use nativelink_store::store_manager::StoreManager; +use nativelink_util::buf_channel::make_buf_channel_pair; use nativelink_util::common::DigestInfo; -use nativelink_util::digest_hasher::make_ctx_for_hash_func; -use nativelink_util::store_trait::{Store, StoreLike}; +use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc, make_ctx_for_hash_func}; +use nativelink_util::store_trait::{Store, StoreLike, UploadSizeInfo}; use opentelemetry::context::FutureExt; +use prost::Message; +use tokio_util::io::StreamReader; use tonic::{Request, Response, Status}; use tracing::{Instrument, Level, debug, error_span, instrument}; +/// Metrics for the experimental `SplitBlob`/`SpliceBlob` chunking RPCs. +/// The split hit rate (`split_hits` / `split_requests_total`) indicates how +/// often chunked downloads could be served; the spliced/split byte totals +/// bound the transfer volume flowing through the chunked paths. +#[derive(Debug, Default)] +pub struct ChunkingMetrics { + /// Total `SpliceBlob` requests received on chunking-enabled instances. + pub splice_requests_total: AtomicU64, + /// `SpliceBlob` requests that were no-ops because the blob and its chunk + /// layout were already registered. + pub splice_already_exists: AtomicU64, + /// `SpliceBlob` requests rejected because the re-assembled blob did not + /// match the expected digest or size. + pub splice_verification_failures: AtomicU64, + /// Total bytes of blobs successfully re-assembled by `SpliceBlob`. + pub splice_bytes_total: AtomicU64, + /// Total `SplitBlob` requests received on chunking-enabled instances. + pub split_requests_total: AtomicU64, + /// `SplitBlob` requests served from a stored chunk layout. + pub split_hits: AtomicU64, + /// `SplitBlob` requests that could not be served because the blob was + /// not present in the CAS. + pub split_misses: AtomicU64, + /// `SplitBlob` requests served by chunking the blob on demand because + /// no stored layout was available (or its chunks were evicted). + pub split_chunked_on_demand: AtomicU64, + /// Total bytes of blobs served as chunk layouts by `SplitBlob`. + pub split_bytes_total: AtomicU64, +} + +impl MetricsComponent for ChunkingMetrics { + fn publish( + &self, + _kind: MetricKind, + field_metadata: MetricFieldData, + ) -> Result { + let _enter = group!(field_metadata.name).entered(); + + publish!( + "splice_requests_total", + &self.splice_requests_total, + MetricKind::Counter, + "Total SpliceBlob requests received" + ); + publish!( + "splice_already_exists", + &self.splice_already_exists, + MetricKind::Counter, + "SpliceBlob requests that were no-ops because blob and layout already existed" + ); + publish!( + "splice_verification_failures", + &self.splice_verification_failures, + MetricKind::Counter, + "SpliceBlob requests rejected due to digest or size mismatch" + ); + publish!( + "splice_bytes_total", + &self.splice_bytes_total, + MetricKind::Counter, + "Total bytes of blobs re-assembled by SpliceBlob" + ); + publish!( + "split_requests_total", + &self.split_requests_total, + MetricKind::Counter, + "Total SplitBlob requests received" + ); + publish!( + "split_hits", + &self.split_hits, + MetricKind::Counter, + "SplitBlob requests served from a stored chunk layout" + ); + publish!( + "split_misses", + &self.split_misses, + MetricKind::Counter, + "SplitBlob requests where the blob was not present" + ); + publish!( + "split_chunked_on_demand", + &self.split_chunked_on_demand, + MetricKind::Counter, + "SplitBlob requests served by chunking the blob on demand" + ); + publish!( + "split_bytes_total", + &self.split_bytes_total, + MetricKind::Counter, + "Total bytes of blobs served as chunk layouts by SplitBlob" + ); + + Ok(MetricPublishKnownKindData::Component) + } +} + +/// Per-instance state for the experimental chunking RPCs. +#[derive(Debug, Clone)] +struct ChunkingInstance { + /// Store holding blob-digest -> chunk-layout mappings. + index_store: Store, + /// Average chunk size used for server-side `FastCDC` 2020 chunking. + avg_chunk_size_bytes: u32, + /// Maximum number of chunks accepted in a `SpliceBlob` request or + /// produced by on-demand chunking. + max_chunk_count: usize, +} + +impl ChunkingInstance { + /// Maximum serialized layout size consistent with `max_chunk_count`. + const fn max_layout_size(&self) -> u64 { + self.max_chunk_count as u64 * MAX_LAYOUT_BYTES_PER_CHUNK + } +} + #[derive(Debug)] pub struct CasServer { stores: HashMap, + chunking_instances: HashMap, + chunking_metrics: ChunkingMetrics, } type GetTreeStream = Pin> + Send + 'static>>; @@ -52,25 +179,109 @@ type GetTreeStream = Pin> /// Per-blob deadline applied inside `BatchReadBlobs` / `BatchUpdateBlobs`. const BATCH_PER_BLOB_TIMEOUT: Duration = Duration::from_secs(30); +/// Maximum size of a single chunk accepted in a `SpliceBlob` request. +/// Deliberately looser than the largest chunk the server ever advertises +/// (4x the maximum allowed average = 4 MiB) so clients using their own +/// chunking function are still accepted. Together with `CHUNK_CONCURRENCY` +/// this bounds the memory a single splice request can pin. +const MAX_SPLICE_CHUNK_SIZE: u64 = 16 * 1024 * 1024; + +/// Generous upper bound for the serialized size of one chunk entry in a +/// stored layout (hash string of up to 128 hex characters plus varints and +/// field tags). Multiplied by the configured `max_chunk_count` this caps +/// layout reads from the index store; a larger entry is corrupt. A truncated +/// read is detected (and treated as no layout) by the size consistency check +/// in `read_chunk_layout`. +const MAX_LAYOUT_BYTES_PER_CHUNK: u64 = 160; + +/// Number of chunk reads/writes kept in flight while re-assembling or +/// chunking a blob. Matches the `DedupStore` concurrency default. +const CHUNK_CONCURRENCY: usize = 10; + impl CasServer { pub fn new( configs: &[WithInstanceName], store_manager: &StoreManager, ) -> Result { let mut stores = HashMap::with_capacity(configs.len()); + let mut chunking_instances = HashMap::new(); for config in configs { let store = store_manager.get_store(&config.cas_store).ok_or_else(|| { make_input_err!("'cas_store': '{}' does not exist", config.cas_store) })?; + if let Some(chunking_config) = &config.experimental_chunking { + let avg_chunk_size_bytes = chunking_config + .validated_avg_chunk_size_bytes() + .err_tip(|| { + format!( + "In 'experimental_chunking' of instance '{}'", + config.instance_name + ) + })?; + if store.downcast_ref::(None).is_some() { + // SplitBlob/SpliceBlob for grpc-store-backed instances + // are forwarded to the backend, which owns the chunk + // layouts; a local index store is meaningless there. + error_if!( + chunking_config.index_store.is_some(), + "'experimental_chunking.index_store' of instance '{}' must not be set when 'cas_store' is a grpc store: SplitBlob/SpliceBlob are forwarded to the backend", + config.instance_name + ); + // No ChunkingInstance: the forwarding shortcut in the + // handlers takes over before local chunking is reached. + stores.insert(config.instance_name.clone(), store); + continue; + } + let index_store_name = chunking_config.index_store.as_ref().ok_or_else(|| { + make_input_err!( + "'experimental_chunking.index_store' of instance '{}' is required", + config.instance_name + ) + })?; + // Chunk layouts are stored under the digests of the blobs + // they describe but do not hash to them, so writing them + // into the CAS itself would overwrite blob content. + error_if!( + index_store_name == &config.cas_store, + "'experimental_chunking.index_store' of instance '{}' must not be the same store as 'cas_store'", + config.instance_name + ); + let index_store = store_manager.get_store(index_store_name).ok_or_else(|| { + make_input_err!( + "'experimental_chunking.index_store': '{index_store_name}' does not exist" + ) + })?; + let avg_chunk_size_bytes = u32::try_from(avg_chunk_size_bytes) + .err_tip(|| "avg_chunk_size_bytes did not fit in u32")?; + let max_chunk_count = usize::try_from(chunking_config.resolved_max_chunk_count()) + .err_tip(|| "max_chunk_count did not fit in usize")?; + chunking_instances.insert( + config.instance_name.clone(), + ChunkingInstance { + index_store, + avg_chunk_size_bytes, + max_chunk_count, + }, + ); + } stores.insert(config.instance_name.clone(), store); } - Ok(Self { stores }) + Ok(Self { + stores, + chunking_instances, + chunking_metrics: ChunkingMetrics::default(), + }) } pub fn into_service(self) -> Server { Server::new(self) } + /// Metrics for the experimental `SplitBlob`/`SpliceBlob` RPCs. + pub const fn chunking_metrics(&self) -> &ChunkingMetrics { + &self.chunking_metrics + } + async fn inner_find_missing_blobs( &self, request: FindMissingBlobsRequest, @@ -330,6 +541,536 @@ impl CasServer { }) .right_stream()) } + + /// Returns the CAS store for an instance and its chunking state, or + /// `Unimplemented` when chunking is not enabled for it. Grpc-store-backed + /// instances never reach this: their handlers forward the RPC to the + /// backend first. + fn chunking_instance(&self, instance_name: &str) -> Result<(Store, ChunkingInstance), Error> { + let store = self + .stores + .get(instance_name) + .err_tip(|| format!("'instance_name' not configured for '{instance_name}'"))? + .clone(); + let chunking_instance = self + .chunking_instances + .get(instance_name) + .ok_or_else(|| { + make_err!( + Code::Unimplemented, + "Blob chunking is not enabled for instance '{instance_name}'" + ) + })? + .clone(); + Ok((store, chunking_instance)) + } + + /// Returns the backend `GrpcStore` when the instance's CAS is a grpc + /// proxy store, in which case chunking RPCs are forwarded verbatim. + fn grpc_store_for_instance(&self, instance_name: &str) -> Option<&GrpcStore> { + self.stores + .get(instance_name) + .and_then(|store| store.downcast_ref::(None)) + } + + /// Returns the digest function explicitly requested by the client, or + /// `None` when the field was left unset. REAPI's length-based inference + /// cannot be used as a fallback here: SHA256 and BLAKE3 digests are both + /// 32 bytes, and `NativeLink` announces support for both. Notably Bazel + /// (9.1.1) leaves this field unset even when running with + /// `--digest_function=blake3`. + fn explicit_hasher_func(digest_function_value: i32) -> Option { + digest_function::Value::try_from(digest_function_value) + .ok() + .and_then(|value| DigestHasherFunc::try_from(value).ok()) + } + + /// Determines the digest function of a blob already present in the CAS + /// by hashing its content with each supported function and returning the + /// one that reproduces `blob_digest`. + async fn infer_blob_hasher_func( + store: &Store, + blob_digest: DigestInfo, + ) -> Result { + const CANDIDATES: [DigestHasherFunc; 2] = + [DigestHasherFunc::Sha256, DigestHasherFunc::Blake3]; + let (tx, rx) = make_buf_channel_pair(); + let read_store = store.clone(); + let read_fut = async move { + let mut tx = tx; + read_store + .get_part(blob_digest, &mut tx, 0, None) + .await + .err_tip(|| "Failed to read blob in infer_blob_hasher_func") + }; + let hash_fut = async move { + let mut rx = rx; + let mut hashers = CANDIDATES.map(|func| func.hasher()); + loop { + let data = rx + .recv() + .await + .err_tip(|| "In infer_blob_hasher_func::recv")?; + if data.is_empty() { + break; // EOF. + } + for hasher in &mut hashers { + hasher.update(&data); + } + } + Ok::<_, Error>(hashers.map(|mut hasher| hasher.finalize_digest())) + }; + let (read_res, hash_res) = futures::join!(read_fut, hash_fut); + let computed_digests = read_res.merge(hash_res)?; + CANDIDATES + .iter() + .zip(computed_digests) + .find(|(_, computed)| *computed == blob_digest) + .map(|(func, _)| *func) + .ok_or_else(|| { + make_err!( + Code::NotFound, + "Blob {blob_digest} does not match any supported digest function; no split information available" + ) + }) + } + + /// Returns the display names of the chunks missing from the CAS. The + /// existence check also touches present chunks, which extends their + /// lifetimes on a best-effort basis (stores that answer existence from a + /// cache may not promote the underlying entries). + async fn missing_chunks(store: &Store, chunk_digests: &[Digest]) -> Result, Error> { + let mut digest_infos = Vec::with_capacity(chunk_digests.len()); + for digest in chunk_digests { + digest_infos + .push(DigestInfo::try_from(digest.clone()).err_tip(|| "Invalid chunk digest")?); + } + let chunk_keys: Vec<_> = digest_infos.iter().map(|digest| (*digest).into()).collect(); + let sizes = store + .has_many(&chunk_keys) + .await + .err_tip(|| "In missing_chunks")?; + Ok(sizes + .iter() + .zip(&digest_infos) + .filter(|(maybe_size, _)| maybe_size.is_none()) + .map(|(_, digest)| digest.to_string()) + .collect()) + } + + /// Reads the chunk layout registered for a blob. Returns `None` when no + /// usable layout exists: not registered, undecodable, or inconsistent + /// with the blob size (which also rejects entries truncated by the read + /// cap below). + async fn read_chunk_layout( + chunking_instance: &ChunkingInstance, + blob_digest: DigestInfo, + ) -> Option { + let layout_bytes = chunking_instance + .index_store + .get_part_unchunked(blob_digest, 0, Some(chunking_instance.max_layout_size())) + .await + .ok()?; + let layout = SplitBlobResponse::decode(layout_bytes).ok()?; + // A usable layout must reproduce the blob exactly, so the chunk + // sizes have to add up to the blob size. + let mut total_size: u64 = 0; + for digest in &layout.chunk_digests { + total_size = total_size.checked_add(u64::try_from(digest.size_bytes).ok()?)?; + } + (total_size == blob_digest.size_bytes()).then_some(layout) + } + + /// Writes the chunk layout for a blob to the index store. This is the + /// write side of the format `read_chunk_layout` expects. + async fn write_chunk_layout( + index_store: &Store, + blob_digest: DigestInfo, + layout: &SplitBlobResponse, + ) -> Result<(), Error> { + index_store + .update_oneshot(blob_digest, layout.encode_to_vec().into()) + .await + .err_tip(|| "Failed to write chunk layout to index store") + } + + async fn inner_split_blob( + &self, + request: SplitBlobRequest, + ) -> Result, Error> { + // If we are a GrpcStore we forward the RPC to the backend, which + // owns chunking and the layout index for proxied instances. + if let Some(grpc_store) = self.grpc_store_for_instance(&request.instance_name) { + return grpc_store.split_blob(Request::new(request)).await; + } + let (store, chunking_instance) = self.chunking_instance(&request.instance_name)?; + self.chunking_metrics + .split_requests_total + .fetch_add(1, Ordering::Relaxed); + + let blob_digest: DigestInfo = request + .blob_digest + .err_tip(|| "Expected blob_digest to exist in SplitBlobRequest")? + .try_into() + .err_tip(|| "In SplitBlobRequest::blob_digest")?; + + // The existence check also touches the blob, extending its lifetime + // (best effort) as suggested by the REAPI spec for SplitBlob. + let (blob_exists, maybe_layout) = futures::join!( + store.has(blob_digest), + Self::read_chunk_layout(&chunking_instance, blob_digest), + ); + if blob_exists.err_tip(|| "In split_blob")?.is_none() { + self.chunking_metrics + .split_misses + .fetch_add(1, Ordering::Relaxed); + return Err(make_err!( + Code::NotFound, + "Blob {blob_digest} not present in the CAS in split_blob" + )); + } + + // Serve the registered layout if it is still fully backed by chunks + // in the CAS. Any problem with it (missing, corrupt, evicted chunks, + // or a transient chunk existence-check failure) falls back to + // re-chunking the blob below. + if let Some(layout) = maybe_layout + && matches!( + Self::missing_chunks(&store, &layout.chunk_digests).await, + Ok(missing) if missing.is_empty() + ) + { + self.chunking_metrics + .split_hits + .fetch_add(1, Ordering::Relaxed); + self.chunking_metrics + .split_bytes_total + .fetch_add(blob_digest.size_bytes(), Ordering::Relaxed); + return Ok(Response::new(layout)); + } + + // No usable layout: chunk the blob on demand with FastCDC 2020, + // store the chunks and the layout, and serve the result. This is the + // path taken for blobs that were uploaded whole (e.g. outputs + // produced by remote execution workers). + let split_response = self + .chunk_blob_on_demand( + &store, + &chunking_instance, + blob_digest, + request.digest_function, + ) + .await?; + self.chunking_metrics + .split_chunked_on_demand + .fetch_add(1, Ordering::Relaxed); + self.chunking_metrics + .split_bytes_total + .fetch_add(blob_digest.size_bytes(), Ordering::Relaxed); + Ok(Response::new(split_response)) + } + + /// Chunks the blob with `FastCDC` 2020 (normalization level 2, parameters + /// derived from the configured average chunk size per the REAPI spec), + /// uploads any missing chunks to the CAS, registers the layout in the + /// index store, and returns it. + async fn chunk_blob_on_demand( + &self, + store: &Store, + chunking_instance: &ChunkingInstance, + blob_digest: DigestInfo, + digest_function_value: i32, + ) -> Result { + let avg_size = chunking_instance.avg_chunk_size_bytes; + let (min_size, max_size) = (avg_size / 4, avg_size * 4); + // Chunk digests MUST use the blob's digest function. When the client + // leaves the field unset it has to be inferred from the blob content + // (an extra read pass) since the hash length alone is ambiguous. + let hasher_func = match Self::explicit_hasher_func(digest_function_value) { + Some(hasher_func) => hasher_func, + None => Self::infer_blob_hasher_func(store, blob_digest).await?, + }; + + let (tx, rx) = make_buf_channel_pair(); + let read_store = store.clone(); + // `tx` is moved into the future so that when the read finishes or + // fails it is dropped, which terminates the chunking stream. + let read_fut = async move { + let mut tx = tx; + read_store + .get_part(blob_digest, &mut tx, 0, None) + .await + .err_tip(|| format!("Failed to read blob {blob_digest} in chunk_blob_on_demand")) + }; + // `rx` is owned by this future so an early error return drops it, + // which aborts the in-flight read instead of leaving it blocked. + let chunk_fut = async move { + let mut bytes_reader = StreamReader::new(rx); + let mut cdc = AsyncStreamCDC::with_level( + &mut bytes_reader, + min_size, + avg_size, + max_size, + Normalization::Level2, + ); + // Chunks are hashed and stored CHUNK_CONCURRENCY at a time while + // the blob keeps streaming; `buffered` preserves chunk order. + let chunk_digests: Vec = pin!(cdc.as_stream()) + .map(|chunk_result| async { + let chunk = chunk_result + .map_err(|e| make_err!(Code::Internal, "Failed to chunk blob: {e:?}")) + .err_tip(|| "In chunk_blob_on_demand")?; + let mut hasher = hasher_func.hasher(); + hasher.update(&chunk.data); + let chunk_digest = hasher.finalize_digest(); + // The existence check also touches pre-existing chunks, + // extending their lifetimes (best effort). FastCDC is + // deterministic, so repeated splits of similar blobs + // mostly find their chunks present. + if store + .has(chunk_digest) + .await + .err_tip(|| "In chunk_blob_on_demand")? + .is_none() + { + store + .update_oneshot(chunk_digest, chunk.data.into()) + .await + .err_tip(|| { + format!( + "Failed to store chunk {chunk_digest} in chunk_blob_on_demand" + ) + })?; + } + Ok::(chunk_digest.into()) + }) + .buffered(CHUNK_CONCURRENCY) + .try_collect() + .await?; + Ok::, Error>(chunk_digests) + }; + let (read_res, chunk_res) = futures::join!(read_fut, chunk_fut); + // Prefer the read error (the chunker error is usually a consequence + // of it); merge keeps both messages when both fail. + let chunk_digests = read_res + .merge(chunk_res) + .err_tip(|| "Failed to chunk blob in chunk_blob_on_demand")?; + if chunk_digests.len() > chunking_instance.max_chunk_count { + return Err(make_err!( + Code::NotFound, + "Blob {blob_digest} produced {} chunks, exceeding the configured max_chunk_count of {}; no split information available", + chunk_digests.len(), + chunking_instance.max_chunk_count + )); + } + + let split_response = SplitBlobResponse { + chunk_digests, + chunking_function: chunking_function::Value::FastCdc2020.into(), + }; + Self::write_chunk_layout(&chunking_instance.index_store, blob_digest, &split_response) + .await?; + Ok(split_response) + } + + async fn inner_splice_blob( + &self, + request: SpliceBlobRequest, + ) -> Result, Error> { + // If we are a GrpcStore we forward the RPC to the backend, which + // owns chunking and the layout index for proxied instances. + if let Some(grpc_store) = self.grpc_store_for_instance(&request.instance_name) { + return grpc_store.splice_blob(Request::new(request)).await; + } + let (store, chunking_instance) = self.chunking_instance(&request.instance_name)?; + let index_store = chunking_instance.index_store; + self.chunking_metrics + .splice_requests_total + .fetch_add(1, Ordering::Relaxed); + + let blob_digest: DigestInfo = request + .blob_digest + .err_tip(|| "Expected blob_digest to exist in SpliceBlobRequest")? + .try_into() + .err_tip(|| "In SpliceBlobRequest::blob_digest")?; + + error_if!( + request.chunk_digests.is_empty(), + "chunk_digests must not be empty in splice_blob" + ); + error_if!( + request.chunk_digests.len() > chunking_instance.max_chunk_count, + "Request has {} chunk_digests, expected at most {} in splice_blob", + request.chunk_digests.len(), + chunking_instance.max_chunk_count + ); + let mut chunk_digests = Vec::with_capacity(request.chunk_digests.len()); + let mut total_size: u64 = 0; + for digest in &request.chunk_digests { + let digest_info = DigestInfo::try_from(digest.clone()) + .err_tip(|| "In SpliceBlobRequest::chunk_digests")?; + error_if!( + digest_info.size_bytes() == 0 || digest_info.size_bytes() > MAX_SPLICE_CHUNK_SIZE, + "Chunk {digest_info} has invalid size, expected to be in range (0, {MAX_SPLICE_CHUNK_SIZE}] in splice_blob" + ); + total_size += digest_info.size_bytes(); + chunk_digests.push(digest_info); + } + if total_size != blob_digest.size_bytes() { + self.chunking_metrics + .splice_verification_failures + .fetch_add(1, Ordering::Relaxed); + return Err(make_err!( + Code::InvalidArgument, + "Sum of chunk sizes ({total_size}) does not match the expected blob size ({}) in splice_blob", + blob_digest.size_bytes() + )); + } + + // One round of existence checks: the chunks (which also touches + // them, best-effort extending their lifetimes), the blob, and the + // registered layout. + let (missing_chunks, blob_exists, layout_exists) = futures::join!( + Self::missing_chunks(&store, &request.chunk_digests), + store.has(blob_digest), + index_store.has(blob_digest), + ); + let missing_chunks = missing_chunks.err_tip(|| "In splice_blob")?; + if !missing_chunks.is_empty() { + return Err(make_err!( + Code::NotFound, + "Chunk(s) [{}] not present in the CAS in splice_blob", + missing_chunks.join(", ") + )); + } + // Fast path: if the blob and its chunk layout are already registered + // this request is a no-op. + if blob_exists.err_tip(|| "In splice_blob")?.is_some() + && layout_exists.err_tip(|| "In splice_blob")?.is_some() + { + self.chunking_metrics + .splice_already_exists + .fetch_add(1, Ordering::Relaxed); + return Ok(Response::new(SpliceBlobResponse { + blob_digest: Some(blob_digest.into()), + })); + } + + // Re-assemble the blob into the store: chunk reads are pipelined + // CHUNK_CONCURRENCY at a time while hashing and channel writes stay + // in chunk order. The digest is verified before the final EOF is + // sent, so a digest mismatch aborts the upload before the store + // commits it. + // When the client sets the digest function, verify with exactly that + // function. When it is unset the hash length is ambiguous (SHA256 + // and BLAKE3 are both 32 bytes), so hash with both candidates and + // accept whichever reproduces the expected digest. + let candidate_hasher_funcs: Vec = + match Self::explicit_hasher_func(request.digest_function) { + Some(hasher_func) => vec![hasher_func], + None => vec![DigestHasherFunc::Sha256, DigestHasherFunc::Blake3], + }; + let verification_failed = AtomicBool::new(false); + let verification_failed_ref = &verification_failed; + let (tx, rx) = make_buf_channel_pair(); + let send_store = store.clone(); + // `tx` is moved into the future so that an early error return drops + // it without an EOF, which aborts the in-flight store update instead + // of leaving it waiting for more data. + let send_fut = async move { + let mut tx = tx; + let mut hashers: Vec<_> = candidate_hasher_funcs + .iter() + .map(DigestHasherFunc::hasher) + .collect(); + let mut fetch_stream = futures::stream::iter(chunk_digests.into_iter().map( + move |chunk_digest| { + let store = send_store.clone(); + async move { + let data = store + .get_part_unchunked(chunk_digest, 0, None) + .await + .err_tip(|| { + format!("Failed to read chunk {chunk_digest} in splice_blob") + })?; + if u64::try_from(data.len()).unwrap_or(0) != chunk_digest.size_bytes() { + return Err(make_err!( + Code::Internal, + "Chunk {chunk_digest} content has length {}, expected {}, in splice_blob", + data.len(), + chunk_digest.size_bytes() + )); + } + Ok::(data) + } + }, + )) + .buffered(CHUNK_CONCURRENCY); + while let Some(data) = fetch_stream.next().await { + let data = data?; + for hasher in &mut hashers { + hasher.update(&data); + } + tx.send(data) + .await + .err_tip(|| "Failed to send chunk data in splice_blob")?; + } + drop(fetch_stream); + let computed_digests: Vec = hashers + .iter_mut() + .map(DigestHasher::finalize_digest) + .collect(); + if !computed_digests + .iter() + .any(|computed| *computed == blob_digest) + { + verification_failed_ref.store(true, Ordering::Relaxed); + return Err(make_err!( + Code::InvalidArgument, + "Digest of spliced blob ({}) does not match the expected digest ({blob_digest}) in splice_blob", + computed_digests + .iter() + .map(ToString::to_string) + .collect::>() + .join(" / ") + )); + } + tx.send_eof() + .err_tip(|| "Failed to send EOF in splice_blob")?; + Ok::<(), Error>(()) + }; + let update_fut = store.update( + blob_digest, + rx, + UploadSizeInfo::ExactSize(blob_digest.size_bytes()), + ); + let (send_res, update_res) = futures::join!(send_fut, update_fut); + if verification_failed.load(Ordering::Relaxed) { + self.chunking_metrics + .splice_verification_failures + .fetch_add(1, Ordering::Relaxed); + } + // Prefer the sender error: it carries the reason the upload was + // aborted (e.g. the digest mismatch), the store error is usually a + // consequence; merge keeps both messages when both fail. + send_res + .merge(update_res) + .err_tip(|| "Failed to write spliced blob to store in splice_blob")?; + + // Persist the chunk layout so SplitBlob can serve it later. + let split_response = SplitBlobResponse { + chunk_digests: request.chunk_digests, + chunking_function: request.chunking_function, + }; + Self::write_chunk_layout(&index_store, blob_digest, &split_response).await?; + + self.chunking_metrics + .splice_bytes_total + .fetch_add(blob_digest.size_bytes(), Ordering::Relaxed); + Ok(Response::new(SpliceBlobResponse { + blob_digest: Some(blob_digest.into()), + })) + } } #[tonic::async_trait] @@ -443,4 +1184,59 @@ impl ContentAddressableStorage for CasServer { } resp } + + #[instrument( + err, + ret(level = Level::DEBUG), + level = Level::ERROR, + skip_all, + fields( + request.instance_name = ?grpc_request.get_ref().instance_name, + request.blob_digest = ?grpc_request.get_ref().blob_digest, + request.digest_function = ?grpc_request.get_ref().digest_function, + ) + )] + async fn split_blob( + &self, + grpc_request: Request, + ) -> Result, Status> { + let request = grpc_request.into_inner(); + let digest_function = request.digest_function; + self.inner_split_blob(request) + .instrument(error_span!("cas_server_split_blob")) + .with_context( + make_ctx_for_hash_func(digest_function).err_tip(|| "In CasServer::split_blob")?, + ) + .await + .err_tip(|| "Failed on split_blob() command") + .map_err(Into::into) + } + + #[instrument( + err, + ret(level = Level::DEBUG), + level = Level::ERROR, + skip_all, + fields( + // Skip request.chunk_digests which is sometimes enormous. + request.instance_name = ?grpc_request.get_ref().instance_name, + request.blob_digest = ?grpc_request.get_ref().blob_digest, + request.digest_function = ?grpc_request.get_ref().digest_function, + ) + )] + async fn splice_blob( + &self, + grpc_request: Request, + ) -> Result, Status> { + let request = grpc_request.into_inner(); + let digest_function = request.digest_function; + self.inner_splice_blob(request) + .instrument(error_span!("cas_server_splice_blob")) + .with_context( + make_ctx_for_hash_func(digest_function).err_tip(|| "In CasServer::splice_blob")?, + ) + .await + .err_tip(|| "Failed on splice_blob() command") + .map_err(Into::into) + } } diff --git a/nativelink-service/tests/cas_server_test.rs b/nativelink-service/tests/cas_server_test.rs index dccc90208..2e1b0c7af 100644 --- a/nativelink-service/tests/cas_server_test.rs +++ b/nativelink-service/tests/cas_server_test.rs @@ -13,6 +13,7 @@ // limitations under the License. use core::pin::Pin; +use core::sync::atomic::Ordering; use core::time::Duration; use std::sync::Arc; @@ -27,8 +28,9 @@ use nativelink_proto::build::bazel::remote::execution::v2::content_addressable_s use nativelink_proto::build::bazel::remote::execution::v2::{ BatchReadBlobsRequest, BatchReadBlobsResponse, BatchUpdateBlobsRequest, BatchUpdateBlobsResponse, Digest, Directory, DirectoryNode, FindMissingBlobsRequest, - GetTreeRequest, GetTreeResponse, NodeProperties, batch_read_blobs_response, - batch_update_blobs_request, batch_update_blobs_response, compressor, digest_function, + GetTreeRequest, GetTreeResponse, NodeProperties, SpliceBlobRequest, SplitBlobRequest, + SplitBlobResponse, batch_read_blobs_response, batch_update_blobs_request, + batch_update_blobs_response, chunking_function, compressor, digest_function, }; use nativelink_proto::google::rpc::Status as GrpcStatus; use nativelink_service::cas_server::CasServer; @@ -37,12 +39,13 @@ use nativelink_store::default_store_factory::store_factory; use nativelink_store::store_manager::StoreManager; use nativelink_util::buf_channel::{DropCloserReadHalf, DropCloserWriteHalf}; use nativelink_util::common::DigestInfo; -use nativelink_util::digest_hasher::DigestHasherFunc; +use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc}; use nativelink_util::health_utils::{HealthStatusIndicator, default_health_status_indicator}; use nativelink_util::store_trait::{ RemoveItemCallback, Store, StoreDriver, StoreKey, StoreLike, UploadSizeInfo, }; use pretty_assertions::assert_eq; +use prost::Message; use prost_types::Timestamp; use tonic::{Code, Request}; @@ -72,6 +75,7 @@ fn make_cas_server(store_manager: &StoreManager) -> Result { instance_name: "foo_instance_name".to_string(), config: nativelink_config::cas_server::CasStoreConfig { cas_store: "main_cas".to_string(), + experimental_chunking: None, }, }], store_manager, @@ -747,6 +751,7 @@ fn make_cas_server_with_stall_store(delay: Duration) -> Result instance_name: INSTANCE_NAME.to_string(), config: nativelink_config::cas_server::CasStoreConfig { cas_store: "main_cas".to_string(), + experimental_chunking: None, }, }], &store_manager, @@ -835,3 +840,708 @@ async fn batch_read_blobs_per_blob_timeout_returns_deadline_exceeded() ); Ok(()) } + +const CHUNK1_VALUE: &str = "hello "; +const CHUNK2_VALUE: &str = "world"; + +async fn make_chunking_store_manager() -> Result, Error> { + let store_manager = make_store_manager().await?; + store_manager.add_store( + "chunk_index", + store_factory( + &StoreSpec::Memory(MemorySpec::default()), + &store_manager, + None, + ) + .await?, + ); + Ok(store_manager) +} + +fn make_chunking_cas_server(store_manager: &StoreManager) -> Result { + make_chunking_cas_server_with_avg(store_manager, 0) +} + +fn make_chunking_cas_server_with_avg( + store_manager: &StoreManager, + avg_chunk_size_bytes: u64, +) -> Result { + CasServer::new( + &[WithInstanceName { + instance_name: INSTANCE_NAME.to_string(), + config: nativelink_config::cas_server::CasStoreConfig { + cas_store: "main_cas".to_string(), + experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig { + index_store: Some("chunk_index".to_string()), + avg_chunk_size_bytes, + max_chunk_count: 0, + }), + }, + }], + store_manager, + ) +} + +/// Uploads the two test chunks to the store and returns their digests and +/// the digest of their concatenation. +async fn upload_test_chunks(store: &Store) -> Result<(Digest, Digest, Digest), Error> { + let chunk1_digest = Digest { + hash: HASH1.to_string(), + size_bytes: CHUNK1_VALUE.len() as i64, + }; + let chunk2_digest = Digest { + hash: HASH2.to_string(), + size_bytes: CHUNK2_VALUE.len() as i64, + }; + store + .update_oneshot( + DigestInfo::try_from(chunk1_digest.clone())?, + CHUNK1_VALUE.into(), + ) + .await?; + store + .update_oneshot( + DigestInfo::try_from(chunk2_digest.clone())?, + CHUNK2_VALUE.into(), + ) + .await?; + let mut hasher = DigestHasherFunc::Sha256.hasher(); + hasher.update(CHUNK1_VALUE.as_bytes()); + hasher.update(CHUNK2_VALUE.as_bytes()); + let blob_digest: Digest = hasher.finalize_digest().into(); + Ok((chunk1_digest, chunk2_digest, blob_digest)) +} + +#[nativelink_test] +async fn splice_and_split_round_trip() -> Result<(), Box> { + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + + let (chunk1_digest, chunk2_digest, blob_digest) = upload_test_chunks(&store).await?; + + let splice_response = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest.clone()), + chunk_digests: vec![chunk1_digest.clone(), chunk2_digest.clone()], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(splice_response.blob_digest.as_ref(), Some(&blob_digest)); + + // The spliced blob must be materialized in the CAS so non-chunking + // clients can read it. + let blob_data = store + .get_part_unchunked(DigestInfo::try_from(blob_digest.clone())?, 0, None) + .await?; + assert_eq!(blob_data, format!("{CHUNK1_VALUE}{CHUNK2_VALUE}")); + + let split_response = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!( + split_response.chunk_digests, + vec![chunk1_digest, chunk2_digest] + ); + assert_eq!( + split_response.chunking_function, + i32::from(chunking_function::Value::FastCdc2020) + ); + + let metrics = cas_server.chunking_metrics(); + assert_eq!(metrics.splice_requests_total.load(Ordering::Relaxed), 1); + assert_eq!( + metrics.splice_bytes_total.load(Ordering::Relaxed), + (CHUNK1_VALUE.len() + CHUNK2_VALUE.len()) as u64 + ); + assert_eq!(metrics.split_requests_total.load(Ordering::Relaxed), 1); + assert_eq!(metrics.split_hits.load(Ordering::Relaxed), 1); + assert_eq!(metrics.split_misses.load(Ordering::Relaxed), 0); + Ok(()) +} + +#[nativelink_test] +async fn splice_blob_rejects_digest_mismatch() -> Result<(), Box> { + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + + let (chunk1_digest, chunk2_digest, _blob_digest) = upload_test_chunks(&store).await?; + let total_size = chunk1_digest.size_bytes + chunk2_digest.size_bytes; + let wrong_blob_digest = Digest { + hash: HASH3.to_string(), + size_bytes: total_size, + }; + + let status = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(wrong_blob_digest.clone()), + chunk_digests: vec![chunk1_digest, chunk2_digest], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::InvalidArgument); + assert!( + status + .message() + .contains("does not match the expected digest"), + "unexpected message: {}", + status.message() + ); + + // The blob must not have been committed to the CAS. + let blob_exists = store.has(DigestInfo::try_from(wrong_blob_digest)?).await?; + assert_eq!(blob_exists, None); + assert_eq!( + cas_server + .chunking_metrics() + .splice_verification_failures + .load(Ordering::Relaxed), + 1 + ); + Ok(()) +} + +#[nativelink_test] +async fn splice_blob_rejects_size_mismatch() -> Result<(), Box> { + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + + let (chunk1_digest, chunk2_digest, blob_digest) = upload_test_chunks(&store).await?; + let wrong_blob_digest = Digest { + size_bytes: blob_digest.size_bytes + 1, + ..blob_digest + }; + + let status = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(wrong_blob_digest), + chunk_digests: vec![chunk1_digest, chunk2_digest], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::InvalidArgument); + assert!( + status + .message() + .contains("does not match the expected blob size"), + "unexpected message: {}", + status.message() + ); + Ok(()) +} + +#[nativelink_test] +async fn splice_blob_missing_chunk_returns_not_found() -> Result<(), Box> { + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + + // Only upload the first chunk. + let chunk1_digest = Digest { + hash: HASH1.to_string(), + size_bytes: CHUNK1_VALUE.len() as i64, + }; + store + .update_oneshot( + DigestInfo::try_from(chunk1_digest.clone())?, + CHUNK1_VALUE.into(), + ) + .await?; + let missing_chunk_digest = Digest { + hash: HASH2.to_string(), + size_bytes: CHUNK2_VALUE.len() as i64, + }; + let mut hasher = DigestHasherFunc::Sha256.hasher(); + hasher.update(CHUNK1_VALUE.as_bytes()); + hasher.update(CHUNK2_VALUE.as_bytes()); + let blob_digest: Digest = hasher.finalize_digest().into(); + + let status = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest), + chunk_digests: vec![chunk1_digest, missing_chunk_digest], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::NotFound); + Ok(()) +} + +#[nativelink_test] +async fn split_blob_absent_blob_returns_not_found() -> Result<(), Box> { + const VALUE: &str = "1"; + + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + + // The blob was never uploaded. + let status = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(Digest { + hash: HASH1.to_string(), + size_bytes: VALUE.len() as i64, + }), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::NotFound); + let metrics = cas_server.chunking_metrics(); + assert_eq!(metrics.split_requests_total.load(Ordering::Relaxed), 1); + assert_eq!(metrics.split_hits.load(Ordering::Relaxed), 0); + assert_eq!(metrics.split_misses.load(Ordering::Relaxed), 1); + Ok(()) +} + +#[nativelink_test] +async fn split_and_splice_disabled_return_unimplemented() -> Result<(), Box> +{ + const VALUE: &str = "1"; + + let store_manager = make_store_manager().await?; + let cas_server = make_cas_server(&store_manager)?; + + let digest = Digest { + hash: HASH1.to_string(), + size_bytes: VALUE.len() as i64, + }; + let split_status = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(digest.clone()), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(split_status.code(), Code::Unimplemented); + + let splice_status = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(digest.clone()), + chunk_digests: vec![digest], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(splice_status.code(), Code::Unimplemented); + Ok(()) +} + +#[nativelink_test] +async fn split_blob_chunks_small_blob_on_demand() -> Result<(), Box> { + const VALUE: &str = "1"; + + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + + // Upload the blob whole (as a remote execution worker would) under its + // real digest, without ever calling SpliceBlob. + let mut hasher = DigestHasherFunc::Sha256.hasher(); + hasher.update(VALUE.as_bytes()); + let blob_digest: Digest = hasher.finalize_digest().into(); + store + .update_oneshot(DigestInfo::try_from(blob_digest.clone())?, VALUE.into()) + .await?; + + let split_response = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest.clone()), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + // A blob smaller than the minimum chunk size is a single chunk whose + // digest equals the blob digest. + assert_eq!(split_response.chunk_digests, vec![blob_digest]); + assert_eq!( + split_response.chunking_function, + i32::from(chunking_function::Value::FastCdc2020) + ); + let metrics = cas_server.chunking_metrics(); + assert_eq!(metrics.split_chunked_on_demand.load(Ordering::Relaxed), 1); + assert_eq!(metrics.split_hits.load(Ordering::Relaxed), 0); + Ok(()) +} + +#[nativelink_test] +async fn split_blob_chunks_large_blob_on_demand_and_reuses_layout() +-> Result<(), Box> { + // Use the smallest allowed average (1 KiB -> min 256, max 4096) so a + // small test blob still produces multiple chunks. + const AVG_CHUNK_SIZE: u64 = 1024; + const BLOB_SIZE: usize = 16 * 1024; + + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server_with_avg(&store_manager, AVG_CHUNK_SIZE)?; + let store = store_manager.get_store("main_cas").unwrap(); + + // Deterministic pseudo-random content so FastCDC finds content-defined + // boundaries. + let mut state = 0x9e37_79b9_u32; + let data: Vec = (0..BLOB_SIZE) + .map(|_| { + state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + (state >> 24) as u8 + }) + .collect(); + let blob_digest = Digest { + hash: HASH1.to_string(), + size_bytes: BLOB_SIZE as i64, + }; + store + .update_oneshot( + DigestInfo::try_from(blob_digest.clone())?, + bytes::Bytes::from(data.clone()), + ) + .await?; + + let split_response = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest.clone()), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert!( + split_response.chunk_digests.len() > 1, + "expected multiple chunks, got {}", + split_response.chunk_digests.len() + ); + + // All chunks must be stored in the CAS and concatenate to the original + // blob in order. + let mut reassembled = Vec::with_capacity(BLOB_SIZE); + for chunk_digest in &split_response.chunk_digests { + let chunk_data = store + .get_part_unchunked(DigestInfo::try_from(chunk_digest.clone())?, 0, None) + .await?; + reassembled.extend_from_slice(&chunk_data); + } + assert_eq!(reassembled, data); + + // A second split must be served from the stored layout. + let second_response = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(second_response.chunk_digests, split_response.chunk_digests); + let metrics = cas_server.chunking_metrics(); + assert_eq!(metrics.split_requests_total.load(Ordering::Relaxed), 2); + assert_eq!(metrics.split_chunked_on_demand.load(Ordering::Relaxed), 1); + assert_eq!(metrics.split_hits.load(Ordering::Relaxed), 1); + Ok(()) +} + +#[nativelink_test] +async fn split_blob_falls_back_when_layout_unusable() -> Result<(), Box> { + const VALUE: &str = "1"; + + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + let index_store = store_manager.get_store("chunk_index").unwrap(); + + let mut hasher = DigestHasherFunc::Sha256.hasher(); + hasher.update(VALUE.as_bytes()); + let blob_digest: Digest = hasher.finalize_digest().into(); + store + .update_oneshot(DigestInfo::try_from(blob_digest.clone())?, VALUE.into()) + .await?; + + // Register a layout whose only chunk is not present in the CAS, + // simulating a chunk that was evicted after the layout was stored. + let stale_layout = SplitBlobResponse { + chunk_digests: vec![Digest { + hash: HASH2.to_string(), + size_bytes: VALUE.len() as i64, + }], + chunking_function: chunking_function::Value::FastCdc2020.into(), + }; + index_store + .update_oneshot( + DigestInfo::try_from(blob_digest.clone())?, + stale_layout.encode_to_vec().into(), + ) + .await?; + + // The unusable layout must be ignored and the blob re-chunked on demand. + let split_response = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest.clone()), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(split_response.chunk_digests, vec![blob_digest]); + let metrics = cas_server.chunking_metrics(); + assert_eq!(metrics.split_hits.load(Ordering::Relaxed), 0); + assert_eq!(metrics.split_chunked_on_demand.load(Ordering::Relaxed), 1); + Ok(()) +} + +#[nativelink_test] +async fn chunking_rejects_index_store_same_as_cas_store() -> Result<(), Box> +{ + let store_manager = make_store_manager().await?; + let error = CasServer::new( + &[WithInstanceName { + instance_name: INSTANCE_NAME.to_string(), + config: nativelink_config::cas_server::CasStoreConfig { + cas_store: "main_cas".to_string(), + experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig { + index_store: Some("main_cas".to_string()), + avg_chunk_size_bytes: 0, + max_chunk_count: 0, + }), + }, + }], + &store_manager, + ) + .err() + .expect("expected same-store index_store to be rejected"); + assert!( + error + .to_string() + .contains("must not be the same store as 'cas_store'"), + "unexpected error: {error}" + ); + Ok(()) +} + +#[nativelink_test] +async fn chunking_on_grpc_store_forbids_index_store() -> Result<(), Box> { + let store_manager = Arc::new(StoreManager::new()); + store_manager.add_store( + "grpc_cas", + store_factory( + &StoreSpec::Grpc(nativelink_config::stores::GrpcSpec { + instance_name: "backend".to_string(), + endpoints: vec![nativelink_config::stores::GrpcEndpoint { + address: "http://localhost:1".to_string(), + tls_config: None, + concurrency_limit: None, + connect_timeout_s: 0, + tcp_keepalive_s: 0, + http2_keepalive_interval_s: 0, + http2_keepalive_timeout_s: 0, + }], + store_type: nativelink_config::stores::StoreType::Cas, + retry: nativelink_config::stores::Retry::default(), + max_concurrent_requests: 0, + connections_per_endpoint: 0, + rpc_timeout_s: 1, + use_legacy_resource_names: false, + headers: std::collections::HashMap::new(), + forward_headers: vec![], + }), + &store_manager, + None, + ) + .await?, + ); + + let make_config = |index_store: Option| { + vec![WithInstanceName { + instance_name: INSTANCE_NAME.to_string(), + config: nativelink_config::cas_server::CasStoreConfig { + cas_store: "grpc_cas".to_string(), + experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig { + index_store, + avg_chunk_size_bytes: 0, + max_chunk_count: 0, + }), + }, + }] + }; + + // A local index store is meaningless when the RPCs are forwarded. + let error = CasServer::new(&make_config(Some("grpc_cas".to_string())), &store_manager) + .err() + .expect("expected index_store on grpc store to be rejected"); + assert!( + error.to_string().contains("must not be set"), + "unexpected error: {error}" + ); + + // Without an index_store the configuration is valid: SplitBlob and + // SpliceBlob are forwarded to the backend. + CasServer::new(&make_config(None), &store_manager)?; + Ok(()) +} + +#[nativelink_test] +async fn max_chunk_count_limits_split_and_splice() -> Result<(), Box> { + // avg 1024 (min allowed) with max_chunk_count 2: the 16 KiB test blob + // chunks to more than 2 pieces, so on-demand splitting must refuse. + const AVG_CHUNK_SIZE: u64 = 1024; + const BLOB_SIZE: usize = 16 * 1024; + + let store_manager = make_chunking_store_manager().await?; + let cas_server = CasServer::new( + &[WithInstanceName { + instance_name: INSTANCE_NAME.to_string(), + config: nativelink_config::cas_server::CasStoreConfig { + cas_store: "main_cas".to_string(), + experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig { + index_store: Some("chunk_index".to_string()), + avg_chunk_size_bytes: AVG_CHUNK_SIZE, + max_chunk_count: 2, + }), + }, + }], + &store_manager, + )?; + let store = store_manager.get_store("main_cas").unwrap(); + + let mut state = 0x9e37_79b9_u32; + let data: Vec = (0..BLOB_SIZE) + .map(|_| { + state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + (state >> 24) as u8 + }) + .collect(); + let blob_digest = Digest { + hash: HASH1.to_string(), + size_bytes: BLOB_SIZE as i64, + }; + store + .update_oneshot( + DigestInfo::try_from(blob_digest.clone())?, + bytes::Bytes::from(data), + ) + .await?; + + let status = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest.clone()), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::NotFound); + assert!( + status.message().contains("max_chunk_count"), + "unexpected message: {}", + status.message() + ); + + // Splices above the cap are rejected outright. + let chunk_digest = Digest { + hash: HASH2.to_string(), + size_bytes: 1, + }; + let status = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest), + chunk_digests: vec![chunk_digest.clone(), chunk_digest.clone(), chunk_digest], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::InvalidArgument); + assert!( + status.message().contains("expected at most 2"), + "unexpected message: {}", + status.message() + ); + Ok(()) +} + +// Bazel 9.1.1 with --digest_function=blake3 leaves digest_function unset in +// SplitBlob/SpliceBlob requests, which is length-ambiguous (SHA256 and +// BLAKE3 are both 32 bytes). The server must infer the function instead of +// assuming the default. +#[nativelink_test] +async fn chunking_infers_blake3_when_digest_function_unset() +-> Result<(), Box> { + const VALUE: &str = "blake3 blob content"; + + let store_manager = make_chunking_store_manager().await?; + let cas_server = make_chunking_cas_server(&store_manager)?; + let store = store_manager.get_store("main_cas").unwrap(); + + let mut hasher = DigestHasherFunc::Blake3.hasher(); + hasher.update(VALUE.as_bytes()); + let blob_digest: Digest = hasher.finalize_digest().into(); + + // Splice: the single chunk is the blob itself, uploaded under its + // BLAKE3 digest, with digest_function left unset. + store + .update_oneshot(DigestInfo::try_from(blob_digest.clone())?, VALUE.into()) + .await?; + let splice_response = cas_server + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(blob_digest.clone()), + chunk_digests: vec![blob_digest.clone()], + digest_function: 0, + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(splice_response.blob_digest.as_ref(), Some(&blob_digest)); + + // On-demand split of a fresh blob uploaded whole: the returned chunk + // digests must be BLAKE3 (here a single chunk equal to the blob). + let mut hasher = DigestHasherFunc::Blake3.hasher(); + hasher.update(b"other blake3 content"); + let other_digest: Digest = hasher.finalize_digest().into(); + store + .update_oneshot( + DigestInfo::try_from(other_digest.clone())?, + bytes::Bytes::from_static(b"other blake3 content"), + ) + .await?; + let split_response = cas_server + .split_blob(Request::new(SplitBlobRequest { + instance_name: INSTANCE_NAME.to_string(), + blob_digest: Some(other_digest.clone()), + digest_function: 0, + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(split_response.chunk_digests, vec![other_digest]); + Ok(()) +} diff --git a/nativelink-service/tests/data/fastcdc2020_test_vectors.txt b/nativelink-service/tests/data/fastcdc2020_test_vectors.txt new file mode 100644 index 000000000..db89efd7d --- /dev/null +++ b/nativelink-service/tests/data/fastcdc2020_test_vectors.txt @@ -0,0 +1,35 @@ +# Test vectors for the FastCDC 2020 content-defined chunking algorithm +# +# Reference implementations: +# - Rust: https://github.com/nlfiedler/fastcdc-rs +# - Go: https://github.com/buildbuddy-io/fastcdc2020 +# +# Test input: +# Image: https://github.com/nlfiedler/fastcdc-rs/blob/49c3d0b/test/fixtures/SekienAkashita.jpg +# SHA256: d9e749d9367fc908876749d6502eb212fee88c9a94892fb07da5ef3ba8bc39ed +# Size: 109466 bytes +# +# Parameters: +# MinSize: 4096 +# AvgSize: 16384 (must be power of 2) +# MaxSize: 65535 +# Normalization: 2 +# +# Format: offset, length, sha256, fingerprint +# The fingerprint is the 64-bit gear hash value at the chunk boundary. + +# Seed: 0 +0 19186 0f9efa589121d5d9e9e2c4ace91337d77cae866537143f6f15a0ffd525a77c2d 17583755766661134474 +19186 19279 c7c86a165573c16448cda35c9169742e85645af42be22889f8b96b8ee0ec7cb0 4098594969649699419 +38465 17354 bc88521e28a8b4479cdea5f75aa721a24f3a0a7d0be903aa6d505c574e51e89d 2365586132076908760 +55819 16387 4b8dac2652e4685c629d2bb1ae9d4448e676b86f2e67ca0b2fff3d9580184b79 16009206469796846404 +72206 19940 c0a7062da6f2386c28e086ee0cedd5732252741269838773cff1ddb05b2df6ed 2473608525189754172 +92146 17320 7fa5b12134dc75cd2ac8dc60d3a8f3c8d22f0ee9d4cf74a4aa937e2a0d2d79a5 2504464741100432583 + +# Seed: 666 +0 17635 cb3a9d80a3569772d4ed331ca37ab0c862c759897b890fc1aac90a4f2ea3a407 17021115692437263050 +17635 17334 d758c6b7b0b7eef1e996f8ccd17de6c645360b03a26c35541e7581348ac08944 8231525949846907466 +34969 19136 24846aefd89e510594bae3e9d7d5ea5012067601512610fed126a3c57ba993f5 10944310959829698982 +54105 17467 efa785e1fefb49f190e665f72fd246c1442079874508c312196da1fb3040d00b 13602876513398592944 +71572 23593 a2f557bdd8d40d8faada963ad5f91ec54b10ccee7c5ae72754a65137592dc607 2945079350535657389 +95165 14301 e131100b4a7147ccad19dc63c4a2fac1f5d8b644e1373eeb6803825024234efc 8981594897574481255 diff --git a/nativelink-service/tests/fastcdc_conformance_test.rs b/nativelink-service/tests/fastcdc_conformance_test.rs new file mode 100644 index 000000000..5d13c5820 --- /dev/null +++ b/nativelink-service/tests/fastcdc_conformance_test.rs @@ -0,0 +1,152 @@ +// Copyright 2024 The NativeLink Authors. All rights reserved. +// +// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// See LICENSE file for details +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Conformance tests for the `FastCDC` 2020 implementation used by the +//! `SplitBlob` on-demand chunking path against the official REAPI test +//! vectors from the remote-apis repository: +//! +//! +//! The chunk boundaries produced by the server MUST match these vectors +//! byte-for-byte; otherwise chunks produced by clients (e.g. Bazel with +//! `--experimental_remote_cache_chunking`) never deduplicate against chunks +//! produced by the server and the feature silently loses its value. + +use fastcdc::v2020::{AsyncStreamCDC, FastCDC, Normalization}; +use futures::StreamExt; +use nativelink_macro::nativelink_test; +use pretty_assertions::assert_eq; +use sha2::{Digest as _, Sha256}; + +/// The canonical test input named in the vectors file header +/// (SHA256 d9e749d9367fc908876749d6502eb212fee88c9a94892fb07da5ef3ba8bc39ed), +/// shared with the existing `DedupStore` `FastCDC` test fixture. +const TEST_INPUT: &[u8] = include_bytes!("../../nativelink-util/tests/data/SekienAkashita.jpg"); +const TEST_VECTORS: &str = include_str!("data/fastcdc2020_test_vectors.txt"); + +/// Parameters stated in the vectors file header. +const MIN_SIZE: u32 = 4096; +const AVG_SIZE: u32 = 16384; +const MAX_SIZE: u32 = 65535; + +struct ExpectedChunk { + offset: u64, + length: usize, + sha256_hex: String, + fingerprint: u64, +} + +/// Parses the `# Seed: ` sections of the vectors file into +/// (seed, expected chunks) pairs. +fn parse_test_vectors() -> Vec<(u64, Vec)> { + let mut sections = Vec::new(); + for line in TEST_VECTORS.lines() { + let line = line.trim(); + if let Some(seed) = line.strip_prefix("# Seed: ") { + sections.push((seed.parse::().unwrap(), Vec::new())); + continue; + } + if line.is_empty() || line.starts_with('#') { + continue; + } + let mut fields = line.split_whitespace(); + let chunk = ExpectedChunk { + offset: fields.next().unwrap().parse().unwrap(), + length: fields.next().unwrap().parse().unwrap(), + sha256_hex: fields.next().unwrap().to_string(), + fingerprint: fields.next().unwrap().parse().unwrap(), + }; + sections + .last_mut() + .expect("chunk line before any '# Seed:' section") + .1 + .push(chunk); + } + assert!(!sections.is_empty(), "no seed sections parsed"); + sections +} + +#[nativelink_test] +async fn fastcdc2020_matches_reapi_test_vectors() -> Result<(), Box> { + assert_eq!( + hex::encode(Sha256::digest(TEST_INPUT)), + "d9e749d9367fc908876749d6502eb212fee88c9a94892fb07da5ef3ba8bc39ed", + "test fixture does not match the input named in the vectors file" + ); + + for (seed, expected_chunks) in parse_test_vectors() { + let chunks: Vec<_> = FastCDC::with_level_and_seed( + TEST_INPUT, + MIN_SIZE, + AVG_SIZE, + MAX_SIZE, + Normalization::Level2, + seed, + ) + .collect(); + assert_eq!( + chunks.len(), + expected_chunks.len(), + "chunk count mismatch for seed {seed}" + ); + for (chunk, expected) in chunks.iter().zip(&expected_chunks) { + assert_eq!(chunk.offset as u64, expected.offset, "offset, seed {seed}"); + assert_eq!(chunk.length, expected.length, "length, seed {seed}"); + assert_eq!(chunk.hash, expected.fingerprint, "fingerprint, seed {seed}"); + let data = &TEST_INPUT[chunk.offset..chunk.offset + chunk.length]; + assert_eq!( + hex::encode(Sha256::digest(data)), + expected.sha256_hex, + "chunk content sha256, seed {seed}" + ); + } + } + Ok(()) +} + +/// The streaming chunker (the variant `SplitBlob` actually uses) must +/// produce the same boundaries as the in-memory reference. +#[nativelink_test] +async fn fastcdc2020_streaming_matches_reapi_test_vectors() +-> Result<(), Box> { + let sections = parse_test_vectors(); + let (_, expected_chunks) = sections + .iter() + .find(|(seed, _)| *seed == 0) + .expect("seed 0 section missing"); + + let mut cdc = AsyncStreamCDC::with_level( + TEST_INPUT, + MIN_SIZE, + AVG_SIZE, + MAX_SIZE, + Normalization::Level2, + ); + let stream = cdc.as_stream(); + let mut stream = core::pin::pin!(stream); + let mut chunks = Vec::new(); + while let Some(chunk) = stream.next().await { + chunks.push(chunk.expect("chunking the test input failed")); + } + assert_eq!(chunks.len(), expected_chunks.len()); + for (chunk, expected) in chunks.iter().zip(expected_chunks) { + assert_eq!(chunk.offset, expected.offset); + assert_eq!(chunk.length, expected.length); + assert_eq!(chunk.hash, expected.fingerprint); + assert_eq!( + hex::encode(Sha256::digest(&chunk.data)), + expected.sha256_hex + ); + } + Ok(()) +} diff --git a/nativelink-store/src/grpc_store.rs b/nativelink-store/src/grpc_store.rs index 40edc4dc0..79d866179 100644 --- a/nativelink-store/src/grpc_store.rs +++ b/nativelink-store/src/grpc_store.rs @@ -29,7 +29,8 @@ use nativelink_proto::build::bazel::remote::execution::v2::content_addressable_s use nativelink_proto::build::bazel::remote::execution::v2::{ ActionResult, BatchReadBlobsRequest, BatchReadBlobsResponse, BatchUpdateBlobsRequest, BatchUpdateBlobsResponse, FindMissingBlobsRequest, FindMissingBlobsResponse, - GetActionResultRequest, GetTreeRequest, GetTreeResponse, UpdateActionResultRequest, + GetActionResultRequest, GetTreeRequest, GetTreeResponse, SpliceBlobRequest, SpliceBlobResponse, + SplitBlobRequest, SplitBlobResponse, UpdateActionResultRequest, }; use nativelink_proto::google::bytestream::byte_stream_client::ByteStreamClient; use nativelink_proto::google::bytestream::{ @@ -335,6 +336,64 @@ impl GrpcStore { .await } + pub async fn split_blob( + &self, + grpc_request: Request, + ) -> Result, Error> { + error_if!( + matches!(self.store_type, nativelink_config::stores::StoreType::Ac), + "CAS operation on AC store" + ); + + let mut request = grpc_request.into_inner(); + request.instance_name.clone_from(&self.instance_name); + self.perform_request(request, |request| async move { + let channel = self + .connection_manager + .connection(format!("split_blob: {:?}", request.blob_digest)) + .await + .err_tip(|| "in split_blob")?; + ContentAddressableStorageClient::new(channel) + .split_blob(enrich_request( + Request::new(request), + &self.headers, + &self.forward_headers, + )) + .await + .err_tip(|| "in GrpcStore::split_blob") + }) + .await + } + + pub async fn splice_blob( + &self, + grpc_request: Request, + ) -> Result, Error> { + error_if!( + matches!(self.store_type, nativelink_config::stores::StoreType::Ac), + "CAS operation on AC store" + ); + + let mut request = grpc_request.into_inner(); + request.instance_name.clone_from(&self.instance_name); + self.perform_request(request, |request| async move { + let channel = self + .connection_manager + .connection(format!("splice_blob: {:?}", request.blob_digest)) + .await + .err_tip(|| "in splice_blob")?; + ContentAddressableStorageClient::new(channel) + .splice_blob(enrich_request( + Request::new(request), + &self.headers, + &self.forward_headers, + )) + .await + .err_tip(|| "in GrpcStore::splice_blob") + }) + .await + } + fn get_read_request(&self, mut request: ReadRequest) -> Result { const IS_UPLOAD_FALSE: bool = false; let mut resource_info = ResourceInfo::new(&request.resource_name, IS_UPLOAD_FALSE)?; diff --git a/nativelink-store/src/verify_store.rs b/nativelink-store/src/verify_store.rs index e3722ff48..1b313e08e 100644 --- a/nativelink-store/src/verify_store.rs +++ b/nativelink-store/src/verify_store.rs @@ -23,13 +23,12 @@ use nativelink_util::buf_channel::{ DropCloserReadHalf, DropCloserWriteHalf, make_buf_channel_pair, }; use nativelink_util::common::PackedHash; -use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc, default_digest_hasher_func}; +use nativelink_util::digest_hasher::{DigestHasher, digest_hasher_func_from_context}; use nativelink_util::health_utils::{HealthStatusIndicator, default_health_status_indicator}; use nativelink_util::metrics_utils::CounterWithTime; use nativelink_util::store_trait::{ RemoveItemCallback, Store, StoreDriver, StoreKey, StoreLike, UploadSizeInfo, }; -use opentelemetry::context::Context; #[derive(Debug, MetricsComponent)] pub struct VerifyStore { @@ -184,12 +183,7 @@ impl StoreDriver for VerifyStore { } let mut hasher = if self.verify_hash { - Some( - Context::current() - .get::() - .map_or_else(default_digest_hasher_func, |v| *v) - .hasher(), - ) + Some(digest_hasher_func_from_context().hasher()) } else { None }; diff --git a/nativelink-store/tests/grpc_store_test.rs b/nativelink-store/tests/grpc_store_test.rs index 3466997b2..858d5a653 100644 --- a/nativelink-store/tests/grpc_store_test.rs +++ b/nativelink-store/tests/grpc_store_test.rs @@ -9,8 +9,14 @@ use futures::{Stream, StreamExt}; use nativelink_config::stores::{GrpcEndpoint, GrpcSpec, Retry, StoreType}; use nativelink_error::{Error, ResultExt}; use nativelink_macro::nativelink_test; +use nativelink_proto::build::bazel::remote::execution::v2::content_addressable_storage_server::{ + ContentAddressableStorage, ContentAddressableStorageServer, +}; use nativelink_proto::build::bazel::remote::execution::v2::{ - FindMissingBlobsRequest, digest_function, + BatchReadBlobsRequest, BatchReadBlobsResponse, BatchUpdateBlobsRequest, + BatchUpdateBlobsResponse, Digest, FindMissingBlobsRequest, FindMissingBlobsResponse, + GetTreeRequest, GetTreeResponse, SpliceBlobRequest, SpliceBlobResponse, SplitBlobRequest, + SplitBlobResponse, chunking_function, digest_function, }; use nativelink_proto::google::bytestream::byte_stream_server::{ByteStream, ByteStreamServer}; use nativelink_proto::google::bytestream::{ @@ -322,3 +328,145 @@ async fn read_works_with_headers() -> Result<(), Error> { Ok(()) } + +#[derive(Debug, Clone)] +struct FakeCasServer { + split_requests: Arc>>, + splice_requests: Arc>>, +} + +impl FakeCasServer { + fn new() -> Self { + Self { + split_requests: Arc::new(Mutex::new(vec![])), + splice_requests: Arc::new(Mutex::new(vec![])), + } + } +} + +type GetTreeStream = Pin> + Send + 'static>>; + +#[tonic::async_trait] +impl ContentAddressableStorage for FakeCasServer { + type GetTreeStream = GetTreeStream; + + #[allow(clippy::unimplemented)] + async fn find_missing_blobs( + &self, + _grpc_request: Request, + ) -> Result, Status> { + unimplemented!(); + } + + #[allow(clippy::unimplemented)] + async fn batch_update_blobs( + &self, + _grpc_request: Request, + ) -> Result, Status> { + unimplemented!(); + } + + #[allow(clippy::unimplemented)] + async fn batch_read_blobs( + &self, + _grpc_request: Request, + ) -> Result, Status> { + unimplemented!(); + } + + #[allow(clippy::unimplemented)] + async fn get_tree( + &self, + _grpc_request: Request, + ) -> Result, Status> { + unimplemented!(); + } + + async fn split_blob( + &self, + grpc_request: Request, + ) -> Result, Status> { + let request = grpc_request.into_inner(); + self.split_requests.lock().await.push(request.clone()); + Ok(Response::new(SplitBlobResponse { + chunk_digests: request.blob_digest.into_iter().collect(), + chunking_function: request.chunking_function, + })) + } + + async fn splice_blob( + &self, + grpc_request: Request, + ) -> Result, Status> { + let request = grpc_request.into_inner(); + self.splice_requests.lock().await.push(request.clone()); + Ok(Response::new(SpliceBlobResponse { + blob_digest: request.blob_digest, + })) + } +} + +async fn make_fake_cas_server() -> (FakeCasServer, u16) { + let fake_cas_server = FakeCasServer::new(); + let server = ContentAddressableStorageServer::new(fake_cas_server.clone()); + let listener = TcpIncoming::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let port = listener.local_addr().unwrap().port(); + + background_spawn!("server", async move { + Server::builder() + .add_service(server) + .serve_with_incoming(listener) + .await + .unwrap(); + }); + + (fake_cas_server, port) +} + +#[nativelink_test] +async fn split_and_splice_blob_forward_to_backend() -> Result<(), Error> { + let (server, port) = make_fake_cas_server().await; + let mut spec = test_spec(format!("http://localhost:{port}"), false); + spec.instance_name = "backend_instance".to_string(); + let store = GrpcStore::new(&spec).await?; + + let digest = Digest { + hash: VALID_HASH.to_string(), + size_bytes: RAW_INPUT.len() as i64, + }; + + let split_response = store + .split_blob(Request::new(SplitBlobRequest { + instance_name: "local_instance".to_string(), + blob_digest: Some(digest.clone()), + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(split_response.chunk_digests, vec![digest.clone()]); + { + let split_requests = server.split_requests.lock().await; + assert_eq!(split_requests.len(), 1); + // The instance name must be rewritten to the backend's. + assert_eq!(split_requests[0].instance_name, "backend_instance"); + } + + let splice_response = store + .splice_blob(Request::new(SpliceBlobRequest { + instance_name: "local_instance".to_string(), + blob_digest: Some(digest.clone()), + chunk_digests: vec![digest.clone()], + digest_function: digest_function::Value::Sha256.into(), + chunking_function: chunking_function::Value::FastCdc2020.into(), + })) + .await? + .into_inner(); + assert_eq!(splice_response.blob_digest, Some(digest)); + { + let splice_requests = server.splice_requests.lock().await; + assert_eq!(splice_requests.len(), 1); + assert_eq!(splice_requests[0].instance_name, "backend_instance"); + } + Ok(()) +} diff --git a/nativelink-util/BUILD.bazel b/nativelink-util/BUILD.bazel index ec442cdcb..f9e0dcd0a 100644 --- a/nativelink-util/BUILD.bazel +++ b/nativelink-util/BUILD.bazel @@ -7,6 +7,9 @@ load( "rust_test_suite", ) +# Shared with the FastCDC conformance test in nativelink-service. +exports_files(["tests/data/SekienAkashita.jpg"]) + rust_library( name = "nativelink-util", srcs = [ diff --git a/nativelink-util/src/digest_hasher.rs b/nativelink-util/src/digest_hasher.rs index 51c911c63..e0a37b0af 100644 --- a/nativelink-util/src/digest_hasher.rs +++ b/nativelink-util/src/digest_hasher.rs @@ -53,6 +53,14 @@ pub fn default_digest_hasher_func() -> DigestHasherFunc { *DEFAULT_DIGEST_HASHER_FUNC.get_or_init(|| DigestHasherFunc::Sha256) } +/// Get the hasher requested by the client from the active context (set via +/// [`make_ctx_for_hash_func`]), falling back to the default hasher. +pub fn digest_hasher_func_from_context() -> DigestHasherFunc { + Context::current() + .get::() + .map_or_else(default_digest_hasher_func, |v| *v) +} + /// Sets the default hasher to use if no hasher was requested by the client. pub fn set_default_digest_hasher_func(hasher: DigestHasherFunc) -> Result<(), Error> { DEFAULT_DIGEST_HASHER_FUNC diff --git a/run_integration_tests.sh b/run_integration_tests.sh index c1009a909..451cf1a27 100755 --- a/run_integration_tests.sh +++ b/run_integration_tests.sh @@ -127,7 +127,10 @@ for pattern in "${TEST_PATTERNS[@]}"; do bazel --output_base="$BAZEL_CACHE_DIR" clean FILENAME=$(basename "$fullpath") echo "Running test $FILENAME" - sudo env RUST_LOG=info docker compose up -d + # sudo resets the environment, so NATIVELINK_DIR must be passed + # through explicitly or docker compose falls back to mounting + # root's ~/.cache/nativelink instead of the per-run cache dir. + sudo env RUST_LOG=info NATIVELINK_DIR="$NATIVELINK_DIR" docker compose up -d if perl -e 'alarm shift; exec @ARGV' 30 bash -c 'until sudo docker compose logs | grep -q "Ready, listening on"; do sleep 1; done'; then echo "String 'Ready, listening on' found in the logs." else diff --git a/src/bin/nativelink.rs b/src/bin/nativelink.rs index 33e1c6ad5..4137940c2 100644 --- a/src/bin/nativelink.rs +++ b/src/bin/nativelink.rs @@ -30,8 +30,8 @@ use hyper_util::server::conn::auto; use hyper_util::service::TowerToHyperService; use mimalloc::MiMalloc; use nativelink_config::cas_server::{ - CasConfig, GlobalConfig, HttpCompressionAlgorithm, ListenerConfig, SchedulerConfig, - ServerConfig, StoreConfig, WorkerConfig, + CasConfig, CasStoreConfig, GlobalConfig, HttpCompressionAlgorithm, ListenerConfig, + SchedulerConfig, ServerConfig, StoreConfig, WithInstanceName, WorkerConfig, }; use nativelink_config::stores::ConfigDigestHashFunction; use nativelink_error::{Code, Error, ResultExt, make_err, make_input_err}; @@ -263,6 +263,17 @@ async fn inner_main( let server_cfgs: Vec = cfg.servers.into_iter().collect(); + // The capabilities service advertises chunking support for CAS instances + // that may be served from a different server block (e.g. behind an L7 + // router), so collect the CAS configs across all blocks. + let all_cas_configs: Vec> = server_cfgs + .iter() + .filter_map(|server_cfg| server_cfg.services.as_ref()) + .filter_map(|services| services.cas.as_deref()) + .flatten() + .cloned() + .collect(); + for server_cfg in server_cfgs { let services = server_cfg .services @@ -292,8 +303,9 @@ async fn inner_main( .add_optional_service( services .cas + .as_deref() .map_or(Ok(None), |cfg| { - CasServer::new(&cfg, &store_manager) + CasServer::new(cfg, &store_manager) .map(|v| Some(service_setup!(v.into_service(), http_config))) }) .err_tip(|| "Could not create CAS service")?, @@ -335,10 +347,9 @@ async fn inner_main( ) .add_optional_service( OptionFuture::from( - services - .capabilities - .as_ref() - .map(|cfg| CapabilitiesServer::new(cfg, &action_schedulers)), + services.capabilities.as_ref().map(|cfg| { + CapabilitiesServer::new(cfg, &action_schedulers, &all_cas_configs) + }), ) .await .map_or(Ok::, Error>(None), |server| { diff --git a/web/apps/docs/content/docs/configuration/chunking.mdx b/web/apps/docs/content/docs/configuration/chunking.mdx new file mode 100644 index 000000000..605284ed7 --- /dev/null +++ b/web/apps/docs/content/docs/configuration/chunking.mdx @@ -0,0 +1,110 @@ +--- +title: Content-defined chunking +description: Cut remote cache transfer bytes by 80-90% for incrementally changing artifacts with the REAPI SplitBlob/SpliceBlob extension and Bazel's --experimental_remote_cache_chunking. +--- + +When a large build output changes slightly — a relinked binary, a container +layer with one file modified — its digest changes, and a conventional remote +cache re-transfers the whole blob. Content-defined chunking (CDC) splits +blobs into chunks at content-derived boundaries, so clients upload and +download only the chunks that actually changed. In our measurements a +one-file change to a 16.8 MB tar layer re-uploaded 2.1 MB (87.5% less), and +a one-source-file change to a linked binary re-uploaded 80.8% less. + +NativeLink implements the server side of the +[REAPI blob split/splice extension](https://github.com/bazelbuild/remote-apis/pull/282): +`SpliceBlob` re-assembles chunked uploads (verifying the digest before +committing anything), and `SplitBlob` serves chunk layouts for downloads — +chunking blobs on demand with FastCDC 2020 when they were uploaded whole, +which is what makes chunked downloads work for outputs produced by remote +execution workers. + +## Requirements + +- **Bazel 9.1.1+ or 8.7.0+** on the client, with + `--experimental_remote_cache_chunking`. Avoid 9.1.0: it has a client bug + that corrupts outputs when the chunking flag is combined with + `--disk_cache` (fixed in 9.1.1). +- Chunking is **optional and off by default**. Without the configuration + below, NativeLink behaves exactly as before and does not advertise + chunking support, so clients fall back to regular transfers. + +## Enabling it + +Add an `experimental_chunking` block to the CAS service and give it a small +store for chunk layouts. The index store must not verify content digests +and must not be the CAS store itself: + +```json5 +stores: [ + { + name: "CAS_MAIN_STORE", + // ... your existing CAS store ... + }, + { + // Blob-to-chunks layouts: roughly 80-140 bytes per chunk. + name: "CHUNK_INDEX_STORE", + filesystem: { + content_path: "/tmp/nativelink/data/content_path-chunk-index", + temp_path: "/tmp/nativelink/data/tmp_path-chunk-index", + eviction_policy: { max_bytes: 100000000 }, + }, + }, +], +servers: [ + { + // ... + services: { + cas: [ + { + cas_store: "CAS_MAIN_STORE", + experimental_chunking: { + index_store: "CHUNK_INDEX_STORE", + // Optional; the REAPI-recommended default. Must be between + // 1 KiB and 1 MiB. Blobs smaller than 4x this value are + // never chunked. + avg_chunk_size_bytes: 524288, + // Optional; blobs producing more chunks than this are served + // without chunking (~25 GiB at the default average). + max_chunk_count: 50000, + }, + }, + ], + // The capabilities service advertises chunking support; clients + // only use it when advertised. + capabilities: [{}], + // ... + }, + }, +], +``` + +A complete runnable example lives at +[`nativelink-config/examples/chunking_cas.json5`](https://github.com/TraceMachina/nativelink/blob/main/nativelink-config/examples/chunking_cas.json5). +For instances whose `cas_store` is a grpc proxy store, omit `index_store`: +the chunking RPCs are forwarded to the backend, which owns the layouts. + +Then build with: + +```sh +bazel build //... \ + --remote_cache=grpc://your-nativelink:50051 \ + --experimental_remote_cache_chunking +``` + +## When it helps, and when it doesn't + +Chunking pays off when clients reach the cache across a real network (WAN, +metered links, cross-region) and artifacts change incrementally: +uncompressed archives, linked binaries, and container layers typically save +80–90% of transfer bytes per change. It does little on same-rack links — +saved bytes only save time when the wire is the bottleneck — and little for +compressed artifacts, where everything after the first changed byte +re-transfers. Small blobs (below 4x the average chunk size) are never +chunked, so hot small-object traffic is unaffected. + +The server verifies every spliced blob's digest before committing it and +materializes the full blob, so non-chunking clients and every existing read +path see ordinary blobs. Storage grows by roughly the chunk bytes for +chunk-eligible blobs; pairing the CAS with a `dedup` or `compression` store +composes normally. diff --git a/web/apps/docs/content/docs/configuration/meta.json b/web/apps/docs/content/docs/configuration/meta.json index 3f037a30c..73c150728 100644 --- a/web/apps/docs/content/docs/configuration/meta.json +++ b/web/apps/docs/content/docs/configuration/meta.json @@ -2,7 +2,8 @@ "pages": [ "intro", "basic", - "production" + "production", + "chunking" ], "title": "Configuring NativeLink" }