/build/source/nativelink-config/src/cas_server.rs

Source
// 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.

use std::collections::HashMap;

use nativelink_error::{Code, Error, ResultExt, make_err};
#[cfg(feature = "dev-schema")]
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use tracing::warn;

use crate::schedulers::SchedulerSpec;
use crate::serde_utils::{
    convert_boolean_with_shellexpand, convert_data_size_with_shellexpand,
    convert_duration_with_shellexpand, convert_numeric_with_shellexpand,
    convert_optional_numeric_with_shellexpand, convert_optional_string_with_shellexpand,
    convert_string_with_shellexpand, convert_vec_string_with_shellexpand,
};
use crate::stores::{
    ClientTlsConfig, ConfigDigestHashFunction, StoreRefName, StoreSpec, StoreType,
};

/// Name of the scheduler. This type will be used when referencing a
/// scheduler in the `CasConfig::schedulers`'s map key.
pub type SchedulerRefName = String;

/// Used when the config references `instance_name` in the protocol.
pub type InstanceName = String;

#[derive(Debug, Default, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct WithInstanceName<T> {
    /// Used when the config references `instance_name` in the protocol.
    #[serde(default)]
    pub instance_name: InstanceName,
    #[serde(flatten)]
    pub config: T,
}

impl<T> core::ops::Deref for WithInstanceName<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        &self.config
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct NamedConfig<Spec> {
    pub name: String,
    #[serde(flatten)]
    pub spec: Spec,
}

#[derive(Deserialize, Serialize, Debug, Default, Clone, Copy)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub enum HttpCompressionAlgorithm {
    /// No compression.
    #[default]
    None,

    /// Zlib compression.
    Gzip,
}

/// Note: Compressing data in the cloud rarely has a benefit, since most
/// cloud providers have very high bandwidth backplanes. However, for
/// clients not inside the data center, it might be a good idea to
/// compress data to and from the cloud. This will however come at a high
/// CPU and performance cost. If you are making remote execution share the
/// same CAS/AC servers as client's remote cache, you can create multiple
/// services with different compression settings that are served on
/// different ports. Then configure the non-cloud clients to use one port
/// and cloud-clients to use another.
#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct HttpCompressionConfig {
    /// The compression algorithm that the server will use when sending
    /// responses to clients. Enabling this will likely save a lot of
    /// data transfer, but will consume a lot of CPU and add a lot of
    /// latency.
    /// see: <https://github.com/tracemachina/nativelink/issues/109>
    ///
    /// Default: `HttpCompressionAlgorithm::None`
    pub send_compression_algorithm: Option<HttpCompressionAlgorithm>,

    /// The compression algorithm that the server will accept from clients.
    /// The server will broadcast the supported compression algorithms to
    /// clients and the client will choose which compression algorithm to
    /// use. Enabling this will likely save a lot of data transfer, but
    /// will consume a lot of CPU and add a lot of latency.
    /// see: <https://github.com/tracemachina/nativelink/issues/109>
    ///
    /// Default: {no supported compression}
    pub accepted_compression_algorithms: Vec<HttpCompressionAlgorithm>,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct AcStoreConfig {
    /// The store name referenced in the `stores` map in the main config.
    /// This store name referenced here may be reused multiple times.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub ac_store: StoreRefName,

    /// Whether the Action Cache store may be written to, this if set to false
    /// it is only possible to read from the Action Cache.
    #[serde(default)]
    pub read_only: bool,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct CasStoreConfig {
    /// The store name referenced in the `stores` map in the main config.
    /// 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<CasChunkingConfig>,
}

#[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<StoreRefName>,

    /// 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<u64, Error> {
        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)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct CapabilitiesRemoteExecutionConfig {
    /// Scheduler used to configure the capabilities of remote execution.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub scheduler: SchedulerRefName,
}

#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct CapabilitiesConfig {
    /// Configuration for remote execution capabilities.
    /// If not set the capabilities service will inform the client that remote
    /// execution is not supported.
    pub remote_execution: Option<CapabilitiesRemoteExecutionConfig>,

    /// Whether this instance supports Bazel remote cache compression.
    /// When enabled, the capabilities service advertises zstd wire compression
    /// and the ByteStream/CAS services accept REAPI compressed-blobs/zstd data.
    ///
    /// Bazel clients enable this with `--remote_cache_compression`.
    #[serde(
        default,
        skip_serializing_if = "is_default",
        deserialize_with = "convert_boolean_with_shellexpand"
    )]
    pub remote_cache_compression: bool,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct ExecutionConfig {
    /// The store name referenced in the `stores` map in the main config.
    /// This store name referenced here may be reused multiple times.
    /// This value must be a CAS store reference.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub cas_store: StoreRefName,

    /// The scheduler name referenced in the `schedulers` map in the main config.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub scheduler: SchedulerRefName,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct FetchConfig {
    /// The store name referenced in the `stores` map in the main config.
    /// This store name referenced here may be reused multiple times.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub fetch_store: StoreRefName,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct PushConfig {
    /// The store name referenced in the `stores` map in the main config.
    /// This store name referenced here may be reused multiple times.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub push_store: StoreRefName,

    /// Whether the Action Cache store may be written to, this if set to false
    /// it is only possible to read from the Action Cache.
    #[serde(default)]
    pub read_only: bool,
}

// From https://github.com/serde-rs/serde/issues/818#issuecomment-287438544
fn is_default<T: Default + PartialEq>(t: &T) -> bool {
    *t == Default::default()
}

#[derive(Deserialize, Serialize, Debug, Default, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct ByteStreamConfig {
    /// Name of the store in the "stores" configuration.
    pub cas_store: StoreRefName,

    /// Max number of bytes to send on each grpc stream chunk.
    /// According to <https://github.com/grpc/grpc.github.io/issues/371>
    /// 16KiB - 64KiB is optimal.
    ///
    ///
    /// Default: 64KiB
    #[serde(
        default,
        deserialize_with = "convert_data_size_with_shellexpand",
        skip_serializing_if = "is_default"
    )]
    pub max_bytes_per_stream: usize,

    /// In the event a client disconnects while uploading a blob, we will hold
    /// the internal stream open for this many seconds before closing it.
    /// This allows clients that disconnect to reconnect and continue uploading
    /// the same blob.
    ///
    /// Default: 10 seconds
    #[serde(
        default,
        deserialize_with = "convert_duration_with_shellexpand",
        skip_serializing_if = "is_default",
        alias = "persist_stream_on_disconnect_timeout"
    )]
    pub persist_stream_on_disconnect_timeout_s: usize,
}

// Older bytestream config. All fields are as per the newer docs, but this requires
// the hashed `cas_stores` v.s. the WithInstanceName approach. This should _not_ be updated
// with newer fields, and eventually dropped
#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
pub struct OldByteStreamConfig {
    pub cas_stores: HashMap<InstanceName, StoreRefName>,
    #[serde(
        default,
        deserialize_with = "convert_data_size_with_shellexpand",
        skip_serializing_if = "is_default"
    )]
    pub max_bytes_per_stream: usize,
    #[serde(
        default,
        deserialize_with = "convert_data_size_with_shellexpand",
        skip_serializing_if = "is_default"
    )]
    pub max_decoding_message_size: usize,
    #[serde(
        default,
        deserialize_with = "convert_duration_with_shellexpand",
        skip_serializing_if = "is_default",
        alias = "persist_stream_on_disconnect_timeout"
    )]
    pub persist_stream_on_disconnect_timeout_s: usize,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct WorkerApiConfig {
    /// The scheduler name referenced in the `schedulers` map in the main config.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub scheduler: SchedulerRefName,

    /// Disable the periodic sweep that tells workers to kill operations
    /// the scheduler no longer has executing on them (for example because
    /// every client disconnected before the action finished). With the
    /// sweep disabled such orphaned actions run to completion and still
    /// warm the action cache, which can be preferable for deployments
    /// with long or expensive actions whose clients merely retry.
    ///
    /// Default: false (the sweep runs)
    #[serde(default)]
    pub disable_kill_revoked_operations: bool,

    /// How often, in seconds, the scheduler checks for operations that
    /// are still running on a worker but are no longer executing
    /// according to the state manager, and tells the worker to kill
    /// them. Each pass costs one state-manager lookup per running
    /// operation, so store-backed deployments with many concurrent
    /// actions may want a longer interval.
    ///
    /// Default: 5 (0 uses the default)
    #[serde(default)]
    pub kill_revoked_operations_interval_s: u64,
}

#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct AdminConfig {
    /// Path to register the admin API. If path is "/admin", and your
    /// domain is "example.com", you can reach the endpoint with:
    /// <http://example.com/admin>.
    ///
    /// Default: "/admin"
    #[serde(default)]
    pub path: String,
}

#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct HealthConfig {
    /// Path to register the health status check. If path is "/status", and your
    /// domain is "example.com", you can reach the endpoint with:
    /// <http://example.com/status>.
    ///
    /// Default: "/status"
    #[serde(default)]
    pub path: String,

    /// Timeout on health checks. Default: 5s.
    #[serde(default)]
    pub timeout_seconds: u64,
}

#[derive(Deserialize, Serialize, Debug)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct BepConfig {
    /// The store to publish build events to.
    /// The store name referenced in the `stores` map in the main config.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub store: StoreRefName,
}

#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct IdentityHeaderSpec {
    /// The name of the header to look for the identity in.
    /// Default: "x-identity"
    #[serde(default, deserialize_with = "convert_optional_string_with_shellexpand")]
    pub header_name: Option<String>,

    /// If the header is required to be set or fail the request.
    #[serde(default)]
    pub required: bool,
}

#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct OriginEventsPublisherSpec {
    /// The store to publish nativelink events to.
    /// The store name referenced in the `stores` map in the main config.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub store: StoreRefName,
}

#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct OriginEventsSpec {
    /// The publisher configuration for origin events.
    pub publisher: OriginEventsPublisherSpec,

    /// The maximum number of events to queue before applying back pressure.
    /// IMPORTANT: Backpressure causes all clients to slow down significantly.
    /// Zero is default.
    ///
    /// Default: 65536 (zero defaults to this)
    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
    pub max_event_queue_size: usize,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct ServicesConfig {
    /// The Content Addressable Storage (CAS) backend config.
    /// The key is the `instance_name` used in the protocol and the
    /// value is the underlying CAS store config.
    #[serde(
        default,
        deserialize_with = "super::backcompat::opt_vec_with_instance_name"
    )]
    pub cas: Option<Vec<WithInstanceName<CasStoreConfig>>>,

    /// The Action Cache (AC) backend config.
    /// The key is the `instance_name` used in the protocol and the
    /// value is the underlying AC store config.
    #[serde(
        default,
        deserialize_with = "super::backcompat::opt_vec_with_instance_name"
    )]
    pub ac: Option<Vec<WithInstanceName<AcStoreConfig>>>,

    /// Capabilities service is required in order to use most of the
    /// bazel protocol. This service is used to provide the supported
    /// features and versions of this bazel gRPC service.
    #[serde(
        default,
        deserialize_with = "super::backcompat::opt_vec_with_instance_name"
    )]
    pub capabilities: Option<Vec<WithInstanceName<CapabilitiesConfig>>>,

    /// The remote execution service configuration.
    /// NOTE: This service is under development and is currently just a
    /// place holder.
    #[serde(
        default,
        deserialize_with = "super::backcompat::opt_vec_with_instance_name"
    )]
    pub execution: Option<Vec<WithInstanceName<ExecutionConfig>>>,

    /// This is the service used to stream data to and from the CAS.
    /// Bazel's protocol strongly encourages users to use this streaming
    /// interface to interact with the CAS when the data is large.
    #[serde(default, deserialize_with = "super::backcompat::opt_bytestream")]
    pub bytestream: Option<Vec<WithInstanceName<ByteStreamConfig>>>,

    /// These two are collectively the Remote Asset protocol, but it's
    /// defined as two separate services
    #[serde(
        default,
        deserialize_with = "super::backcompat::opt_vec_with_instance_name"
    )]
    pub fetch: Option<Vec<WithInstanceName<FetchConfig>>>,

    #[serde(
        default,
        deserialize_with = "super::backcompat::opt_vec_with_instance_name"
    )]
    pub push: Option<Vec<WithInstanceName<PushConfig>>>,

    /// This is the service used for workers to connect and communicate
    /// through.
    /// NOTE: This service should be served on a different, non-public port.
    /// In other words, `worker_api` configuration should not have any other
    /// services that are served on the same port. Doing so is a security
    /// risk, as workers have a different permission set than a client
    /// that makes the remote execution/cache requests.
    pub worker_api: Option<WorkerApiConfig>,

    /// Experimental - Build Event Protocol (BEP) configuration. This is
    /// the service that will consume build events from the client and
    /// publish them to a store for processing by an external service.
    pub experimental_bep: Option<BepConfig>,

    /// This is the service for any administrative tasks.
    /// It provides a REST API endpoint for administrative purposes.
    pub admin: Option<AdminConfig>,

    /// This is the service for health status check.
    pub health: Option<HealthConfig>,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct TlsConfig {
    /// Path to the certificate file.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub cert_file: String,

    /// Path to the private key file.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub key_file: String,

    /// Path to the certificate authority for mTLS, if client authentication is
    /// required for this endpoint.
    #[serde(default, deserialize_with = "convert_optional_string_with_shellexpand")]
    pub client_ca_file: Option<String>,

    /// Path to the certificate revocation list for mTLS, if client
    /// authentication is required for this endpoint.
    #[serde(default, deserialize_with = "convert_optional_string_with_shellexpand")]
    pub client_crl_file: Option<String>,
}

/// Advanced Http configurations. These are generally should not be set.
/// For documentation on what each of these do, see the hyper documentation:
/// See: <https://docs.rs/hyper/latest/hyper/server/conn/struct.Http.html>
///
/// Note: All of these default to hyper's default values unless otherwise
/// specified.
#[derive(Deserialize, Serialize, Debug, Default, Clone, Copy)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct HttpServerConfig {
    /// Interval to send keep-alive pings via HTTP2.
    /// Note: This is in seconds.
    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub http2_keep_alive_interval: Option<u32>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_max_pending_accept_reset_streams: Option<u32>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_initial_stream_window_size: Option<u32>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_initial_connection_window_size: Option<u32>,

    #[serde(default)]
    pub experimental_http2_adaptive_window: Option<bool>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_max_frame_size: Option<u32>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_max_concurrent_streams: Option<u32>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand",
        alias = "experimental_http2_keep_alive_timeout"
    )]
    pub experimental_http2_keep_alive_timeout_s: Option<u32>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_max_send_buf_size: Option<u32>,

    #[serde(default)]
    pub experimental_http2_enable_connect_protocol: Option<bool>,

    #[serde(
        default,
        deserialize_with = "convert_optional_numeric_with_shellexpand"
    )]
    pub experimental_http2_max_header_list_size: Option<u32>,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub enum ListenerConfig {
    /// Listener for HTTP/HTTPS/HTTP2 sockets.
    Http(HttpListener),
}

#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct HttpListener {
    /// Address to listen on. Example: `127.0.0.1:8080` or `:8080` to listen
    /// to all IPs.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub socket_address: String,

    /// Allow binding `socket_address` before it is assigned locally.
    ///
    /// Default: false
    #[serde(default)]
    pub freebind: bool,

    /// Data transport compression configuration to use for this service.
    #[serde(default)]
    pub compression: HttpCompressionConfig,

    /// Advanced Http server configuration.
    #[serde(default)]
    pub advanced_http: HttpServerConfig,

    /// Maximum number of bytes to decode on each grpc stream chunk.
    /// Default: 4 MiB
    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
    pub max_decoding_message_size: usize,

    /// Tls Configuration for this server.
    /// If not set, the server will not use TLS.
    ///
    /// Default: None
    #[serde(default)]
    pub tls: Option<TlsConfig>,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct ServerConfig {
    /// Name of the server. This is used to help identify the service
    /// for telemetry and logs.
    ///
    /// Default: {index of server in config}
    #[serde(default, deserialize_with = "convert_string_with_shellexpand")]
    pub name: String,

    /// Configuration
    pub listener: ListenerConfig,

    /// Services to attach to server.
    pub services: Option<ServicesConfig>,

    /// The config related to identifying the client.
    /// Default: {see `IdentityHeaderSpec`}
    #[serde(default)]
    pub experimental_identity_header: IdentityHeaderSpec,
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub enum WorkerProperty {
    /// List of static values.
    /// Note: Generally there should only ever be 1 value, but if the platform
    /// property key is `PropertyType::Priority` it may have more than one value.
    #[serde(deserialize_with = "convert_vec_string_with_shellexpand")]
    Values(Vec<String>),

    /// A dynamic configuration. The string will be executed as a command
    /// (not shell) and will be split by "\n" (new line character).
    QueryCmd(String),
}

/// Generic config for an endpoint and associated configs.
#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct EndpointConfig {
    /// URI of the endpoint.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub uri: String,

    /// Timeout in seconds that a request should take.
    /// Default: 5 seconds
    pub timeout: Option<f32>,

    /// The TLS configuration to use to connect to the endpoint.
    pub tls_config: Option<ClientTlsConfig>,
}

#[derive(Copy, Clone, Deserialize, Serialize, Debug, Default)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub enum UploadCacheResultsStrategy {
    /// Only upload action results with an exit code of 0.
    #[default]
    SuccessOnly,

    /// Don't upload any action results.
    Never,

    /// Upload all action results that complete.
    Everything,

    /// Only upload action results that fail.
    FailuresOnly,
}

#[derive(Clone, Deserialize, Serialize, Debug)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub enum EnvironmentSource {
    /// The name of the platform property in the action to get the value from.
    Property(String),

    /// The raw value to set.
    Value(#[serde(deserialize_with = "convert_string_with_shellexpand")] String),

    /// Take the value from the local environment corresponding to the name key
    FromEnvironment,

    /// The max amount of time in milliseconds the command is allowed to run
    /// (requested by the client).
    TimeoutMillis,

    /// A special file path will be provided that can be used to communicate
    /// with the parent process about out-of-band information. This file
    /// will be read after the command has finished executing. Based on the
    /// contents of the file, the behavior of the result may be modified.
    ///
    /// The format of the file contents should be json with the following
    /// schema:
    /// {
    ///   // If set the command will be considered a failure.
    ///   // May be one of the following static strings:
    ///   // "timeout": Will Consider this task to be a timeout.
    ///   "failure": "timeout",
    /// }
    ///
    /// All fields are optional, file does not need to be created and may be
    /// empty.
    SideChannelFile,

    /// A "root" directory for the action. This directory can be used to
    /// store temporary files that are not needed after the action has
    /// completed. This directory will be purged after the action has
    /// completed.
    ///
    /// For example:
    /// If an action writes temporary data to a path but nativelink should
    /// clean up this path after the job has executed, you may create any
    /// directory under the path provided in this variable. A common pattern
    /// would be to use `entrypoint` to set a shell script that reads this
    /// variable, `mkdir $ENV_VAR_NAME/tmp` and `export TMPDIR=$ENV_VAR_NAME/tmp`.
    /// Another example might be to bind-mount the `/tmp` path in a container to
    /// this path in `entrypoint`.
    ActionDirectory,
}

#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct UploadActionResultConfig {
    /// Underlying AC store that the worker will use to publish execution results
    /// into. Objects placed in this store should be reachable from the
    /// scheduler/client-cas after they have finished updating.
    /// Default: {No uploading is done}
    pub ac_store: Option<StoreRefName>,

    /// In which situations should the results be published to the `ac_store`,
    /// if set to `SuccessOnly` then only results with an exit code of 0 will be
    /// uploaded, if set to Everything all completed results will be uploaded.
    ///
    /// Default: `SuccessOnly`
    #[serde(default)]
    pub upload_ac_results_strategy: UploadCacheResultsStrategy,

    /// Store to upload historical results to. This should be a CAS store if set.
    ///
    /// Default: {CAS store of parent}
    pub historical_results_store: Option<StoreRefName>,

    /// In which situations should the results be published to the historical CAS.
    /// The historical CAS is where failures are published. These messages conform
    /// to the CAS key-value lookup format and are always a `HistoricalExecuteResponse`
    /// serialized message.
    ///
    /// Default: `FailuresOnly`
    #[serde(default)]
    pub upload_historical_results_strategy: Option<UploadCacheResultsStrategy>,

    /// Template to use for the `ExecuteResponse.message` property. This message
    /// is attached to the response before it is sent to the client. The following
    /// special variables are supported:
    /// - `digest_function`: Digest function used to calculate the action digest.
    /// - `action_digest_hash`: Action digest hash.
    /// - `action_digest_size`: Action digest size.
    /// - `historical_results_hash`: `HistoricalExecuteResponse` digest hash.
    /// - `historical_results_size`: `HistoricalExecuteResponse` digest size.
    ///
    /// A common use case of this is to provide a link to the web page that
    /// contains more useful information for the user.
    ///
    /// An example that is fully compatible with `bb_browser` is:
    /// <https://example.com/my-instance-name-here/blobs/{digest_function}/action/{action_digest_hash}-{action_digest_size}/>
    ///
    /// Default: "" (no message)
    #[serde(default, deserialize_with = "convert_string_with_shellexpand")]
    pub success_message_template: String,

    /// Same as `success_message_template` but for failure case.
    ///
    /// An example that is fully compatible with `bb_browser` is:
    /// <https://example.com/my-instance-name-here/blobs/{digest_function}/historical_execute_response/{historical_results_hash}-{historical_results_size}/>
    ///
    /// Default: "" (no message)
    #[serde(default, deserialize_with = "convert_string_with_shellexpand")]
    pub failure_message_template: String,
}

#[derive(Deserialize, Serialize, Debug, Default)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct LocalWorkerConfig {
    /// Name of the worker. This is give a more friendly name to a worker for logging
    /// and metric publishing. This is also the prefix of the worker id
    /// (i.e. "{name}{uuidv6}").
    /// Default: {Index position in the workers list}
    #[serde(default, deserialize_with = "convert_string_with_shellexpand")]
    pub name: String,

    /// Endpoint which the worker will connect to the scheduler's `WorkerApiService`.
    pub worker_api_endpoint: EndpointConfig,

    /// The maximum time an action is allowed to run. If a task requests for a timeout
    /// longer than this time limit, the task will be rejected. Value in seconds.
    ///
    /// Default: 20 minutes
    #[serde(
        default,
        deserialize_with = "convert_duration_with_shellexpand",
        alias = "max_action_timeout"
    )]
    pub max_action_timeout_s: usize,

    /// Maximum time allowed for uploading action results to CAS after execution
    /// completes. If upload takes longer than this, the action fails with
    /// `DeadlineExceeded` and may be retried by the scheduler. Value in seconds.
    ///
    /// Default: 10 minutes
    #[serde(
        default,
        deserialize_with = "convert_duration_with_shellexpand",
        alias = "max_upload_timeout"
    )]
    pub max_upload_timeout_s: usize,

    /// Maximum time to wait for action directory cleanup before timing out.
    /// Value in seconds.
    ///
    /// Default: 30 seconds
    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
    pub max_cleanup_wait_s: usize,

    /// Maximum backoff duration for exponential backoff when waiting for cleanup.
    /// Value in milliseconds.
    ///
    /// Default: 500 milliseconds
    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
    pub max_cleanup_backoff_ms: usize,

    /// Maximum number of inflight tasks this worker can cope with.
    ///
    /// Default: 0 (infinite tasks)
    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
    pub max_inflight_tasks: u64,

    /// If timeout is handled in `entrypoint` or another wrapper script.
    /// If set to true `NativeLink` will not honor the timeout the action requested
    /// and instead will always force kill the action after `max_action_timeout`
    /// has been reached. If this is set to false, the smaller value of the action's
    /// timeout and `max_action_timeout` will be used to which `NativeLink` will kill
    /// the action.
    ///
    /// The real timeout can be received via an environment variable set in:
    /// `EnvironmentSource::TimeoutMillis`.
    ///
    /// Example on where this is useful: `entrypoint` launches the action inside
    /// a docker container, but the docker container may need to be downloaded. Thus
    /// the timer should not start until the docker container has started executing
    /// the action. In this case, action will likely be wrapped in another program,
    /// like `timeout` and propagate timeouts via `EnvironmentSource::SideChannelFile`.
    ///
    /// Default: false (`NativeLink` fully handles timeouts)
    #[serde(default)]
    pub timeout_handled_externally: bool,

    /// The command to execute on every execution request. This will be parsed as
    /// a command + arguments (not shell).
    /// Example: "run.sh" and a job with command: "sleep 5" will result in a
    /// command like: "run.sh sleep 5".
    /// Default: {Use the command from the job request}.
    #[serde(default, deserialize_with = "convert_string_with_shellexpand")]
    pub entrypoint: String,

    /// An optional script to run before every action is processed on the worker.
    /// The value should be the full path to the script to execute and will pause
    /// all actions on the worker if it returns an exit code other than 0.
    /// If not set, then the worker will never pause and will continue to accept
    /// jobs according to the scheduler configuration.
    /// This is useful, for example, if the worker should not take any more
    /// actions until there is enough resource available on the machine to
    /// handle them.
    pub experimental_precondition_script: Option<String>,

    /// Underlying CAS store that the worker will use to download CAS artifacts.
    /// This store must be a `FastSlowStore`. The `fast` store must be a
    /// `FileSystemStore` because it will use hardlinks when building out the files
    /// instead of copying the files. The slow store must eventually resolve to the
    /// same store the scheduler/client uses to send job requests.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub cas_fast_slow_store: StoreRefName,

    /// Configuration for uploading action results.
    #[serde(default)]
    pub upload_action_result: UploadActionResultConfig,

    /// The directory work jobs will be executed from. This directory will be fully
    /// managed by the worker service and will be purged on startup.
    /// This directory and the directory referenced in `local_filesystem_store_ref`'s
    /// `stores::FilesystemStore::content_path` must be on the same filesystem.
    /// Hardlinks will be used when placing files that are accessible to the jobs
    /// that are sourced from `local_filesystem_store_ref`'s `content_path`.
    #[serde(deserialize_with = "convert_string_with_shellexpand")]
    pub work_directory: String,

    /// Properties of this worker. This configuration will be sent to the scheduler
    /// and used to tell the scheduler to restrict what should be executed on this
    /// worker.
    pub platform_properties: HashMap<String, WorkerProperty>,

    /// An optional mapping of environment names to set for the execution
    /// as well as those specified in the action itself. If set, will set each
    /// key as an environment variable before executing the job with the value
    /// of the environment variable being the value of the property of the
    /// action being executed of that name or the fixed value.
    pub additional_environment: Option<HashMap<String, EnvironmentSource>>,

    /// Optional directory cache configuration for improving performance by caching
    /// reconstructed input directories and using hardlinks instead of rebuilding
    /// them from CAS for every action.
    /// Default: None (directory cache disabled)
    pub directory_cache: Option<DirectoryCacheConfig>,

    /// Optional and experimental: lease every digest in an active action's
    /// input Merkle closure in the worker's locally eviction-managed CAS
    /// tiers (the `cas_fast_slow_store`'s filesystem-backed stores) until the
    /// action has finished cleanup. This prevents `Lost inputs no longer
    /// available remotely` failures caused by local fast-tier eviction while
    /// inputs are being materialized under cache pressure.
    ///
    /// Trade-off: while leases are held, a local filesystem tier may
    /// temporarily exceed its configured `max_bytes` / `max_count` eviction
    /// limits; normal eviction resumes once the action's leases are released.
    /// Operators should leave headroom on the underlying disk when enabling
    /// this.
    ///
    /// Default: false (eviction behavior is unchanged)
    #[serde(default)]
    pub experimental_active_input_leases: bool,

    /// Whether to use namespaces to isolate the execution. This is only available
    /// on Linux. It is highly recommended as it avoids a number of issues with
    /// zombie processes and also provides additional hermeticity. If explicitly set
    /// to true and it is not supported the worker will exit with an error.
    ///
    /// Note: this will fail for non-privileged Dockerised workers, as workers in
    /// Docker don't have permissions to make a new user namespace. Privileged
    /// containers can do this.
    ///
    /// Default: False.
    pub use_namespaces: Option<bool>,

    /// Whether to use a mount namespace to isolate the worker root. This is only
    /// available on Linux and when `use_namespaces` is true. It is highly recommended
    /// provides additional hermeticity. If explicitly set to true and it is not
    /// supported or `use_namespaces` is not set to true the worker will exit with an
    /// error.
    /// Default: False.
    pub use_mount_namespace: Option<bool>,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct DirectoryCacheConfig {
    /// Maximum number of cached directories.
    /// Default: 1000
    #[serde(default = "default_directory_cache_max_entries")]
    pub max_entries: usize,

    /// Maximum total size in bytes for all cached directories (0 = unlimited).
    /// Default: 10737418240 (10 GB)
    #[serde(
        default = "default_directory_cache_max_size_bytes",
        deserialize_with = "convert_data_size_with_shellexpand"
    )]
    pub max_size_bytes: u64,

    /// Base directory for cache storage. This directory will be managed by
    /// the worker and should be on the same filesystem as `work_directory`.
    /// Default: `{work_directory}/../directory_cache`
    #[serde(default, deserialize_with = "convert_string_with_shellexpand")]
    pub cache_root: String,

    /// Optional and experimental: additionally cache every subdirectory by
    /// its own `Directory` digest, not just the root directory. REAPI Merkle
    /// nodes are content-addressed, so a subtree that is byte-identical
    /// between two different roots has the same digest and can be
    /// materialized with a single hardlink pass instead of being rebuilt
    /// from the CAS. This makes the common "one input file changed out of
    /// thousands" case reuse every unchanged subtree.
    ///
    /// Note: subtree caching multiplies the entry COUNT in the cache (every
    /// distinct subdirectory becomes its own entry, subject to the normal
    /// eviction budgets), so operators enabling this should raise
    /// `max_entries` accordingly (for example 10x). To avoid multiplying the
    /// SIZE accounting too, an entry's recorded size covers only bytes not
    /// owned by a descendant entry, so the cache's size total approximates
    /// unique materialized bytes rather than counting each file once per
    /// ancestor level.
    ///
    /// Trade-off: cold constructions pay roughly one extra hardlink pass per
    /// tree level (measured ~10% cold overhead at depth 3, within run
    /// variance) in exchange for the churn-path reuse above.
    ///
    /// Default: false (only root directories are cached; existing behavior)
    #[serde(default)]
    pub experimental_subtree_caching: bool,

    /// Maximum number of concurrent slow-store fetches across ALL directory
    /// constructions of this cache. This bound protects backing stores from
    /// RPC storms: per-level construction concurrency compounds
    /// multiplicatively across tree levels and concurrent actions.
    ///
    /// Interaction with read coalescing: this semaphore fragments coalesced
    /// batches to at most this many items and serializes fetch waves for
    /// trees with many tiny files. Deployments using
    /// `experimental_read_batching` on the underlying grpc store can raise
    /// this substantially (for example 512), since batching collapses the
    /// RPC count. Must be greater than 0.
    ///
    /// Default: 64
    #[serde(default = "default_directory_cache_max_concurrent_fetches")]
    pub max_concurrent_fetches: usize,
    /// On a directory-cache miss, prefetch every `Directory` proto of the
    /// tree with one logical `GetTree` traversal instead of fetching protos
    /// level by level (which costs one round trip per tree DEPTH). The
    /// [REAPI request contract] permits servers to impose their own limit even
    /// when no page size is specified, and the [REAPI response contract]
    /// requires clients to continue with the returned page token. A paginated
    /// traversal may therefore use multiple RPCs. Only takes effect when the
    /// slow tier is a `grpc` store;
    /// any prefetch failure falls back to the per-level path. Worthwhile when
    /// worker-to-CAS latency is non-trivial and trees are deep; measured 5-34x
    /// on the proto phase at 5-25ms RTT. Note the serving CAS pays the tree walk
    /// against its own backend, so its directory protos should be served
    /// from a fast tier.
    ///
    /// [REAPI request contract]: https://github.com/bazelbuild/remote-apis/blob/becdd8f9ff811df88a22d3eadd6341753d51d167/build/bazel/remote/execution/v2/remote_execution.proto#L1932-L1942
    /// [REAPI response contract]: https://github.com/bazelbuild/remote-apis/blob/becdd8f9ff811df88a22d3eadd6341753d51d167/build/bazel/remote/execution/v2/remote_execution.proto#L1961-L1965
    /// Default: false
    #[serde(default)]
    pub experimental_get_tree_prefetch: bool,
}

const fn default_directory_cache_max_entries() -> usize {
    1000
}

const fn default_directory_cache_max_concurrent_fetches() -> usize {
    64
}

const fn default_directory_cache_max_size_bytes() -> u64 {
    10 * 1024 * 1024 * 1024 // 10 GB
}

#[derive(Deserialize, Serialize, Debug)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub enum WorkerConfig {
    /// A worker type that executes jobs locally on this machine.
    Local(LocalWorkerConfig),
}

#[derive(Deserialize, Serialize, Debug, Clone, Copy)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct GlobalConfig {
    /// Maximum number of open files that can be opened at one time.
    /// This value is not strictly enforced, it is a best effort. Some internal libraries
    /// open files or read metadata from a files which do not obey this limit, however
    /// the vast majority of cases will have this limit be honored.
    /// This value must be larger than `ulimit -n` to have any effect.
    /// Any network open file descriptors is not counted in this limit, but is counted
    /// in the kernel limit. It is a good idea to set a very large `ulimit -n`.
    /// Note: This value must be greater than 10.
    ///
    /// Default: 24576 (= 24 * 1024)
    #[serde(deserialize_with = "convert_numeric_with_shellexpand")]
    pub max_open_files: usize,

    /// Default hash function to use while uploading blobs to the CAS when not set
    /// by client.
    ///
    /// Default: `ConfigDigestHashFunction::sha256`
    pub default_digest_hash_function: Option<ConfigDigestHashFunction>,

    /// Default digest size to use for health check when running
    /// diagnostics checks. Health checks are expected to use this
    /// size for filling a buffer that is used for creation of
    /// digest.
    ///
    /// Default: 1024*1024 (1MiB)
    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
    pub default_digest_size_health_check: usize,
}

pub type StoreConfig = NamedConfig<StoreSpec>;
pub type SchedulerConfig = NamedConfig<SchedulerSpec>;

#[derive(Deserialize, Serialize, Debug)]
#[serde(deny_unknown_fields)]
#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
pub struct CasConfig {
    /// List of stores available to use in this config.
    /// The keys can be used in other configs when needing to reference a store.
    pub stores: Vec<StoreConfig>,

    /// Worker configurations used to execute jobs.
    pub workers: Option<Vec<WorkerConfig>>,

    /// List of schedulers available to use in this config.
    /// The keys can be used in other configs when needing to reference a
    /// scheduler.
    pub schedulers: Option<Vec<SchedulerConfig>>,

    /// Servers to setup for this process.
    pub servers: Vec<ServerConfig>,

    /// Experimental - Origin events configuration. This is the service that will
    /// collect and publish nativelink events to a store for processing by an
    /// external service.
    pub experimental_origin_events: Option<OriginEventsSpec>,

    /// Any global configurations that apply to all modules live here.
    pub global: Option<GlobalConfig>,
}

impl CasConfig {
    const ZSTD_COMPRESSION_DOCS_URL: &'static str =
        "https://docs.nativelink.com/configuration/compression";

    /// # Errors
    ///
    /// Will return `Err` if we can't load the file.
    pub fn try_from_json5_file(config_file: &str) -> Result<Self, Error> {
        let json_contents = std::fs::read_to_string(config_file)
            .err_tip(|| format!0("Could not open config file {config_file}"))?0;
        Self::try_from_json5_str(&json_contents)
    }

    fn try_from_json5_str(json_contents: &str) -> Result<Self, Error> {
        let mut config: Self = serde_json5::from_str(json_contents)?0;
        for server in &config.servers24 {
            if let Some(services) = &server.services {
                Self::check_store_conflict(services)?1;
            }
        }
        config.apply_zstd_grpc_store_defaults();
        Ok(config)
    }

    fn zstd_wire_compression_enabled_anywhere(&self) -> bool {
        if self
            .servers
            .iter()
            .filter_map23(|server| server.services.as_ref())
            .filter_map23(|services| services.capabilities.as_deref())
            .flatten()
            .any(|capabilities| capabilities19.remote_cache_compression)
        {
            return true;
        }

        self.stores.iter()22.any22(|store| {
            let mut enabled = false;
            store.spec.visit_grpc_specs(&mut |grpc| {5
                enabled |= matches!2(grpc.store_type, StoreType::Cas)
                    && grpc.experimental_remote_cache_compression == Some(true);
            });
            enabled
        })
    }

    fn apply_zstd_grpc_store_defaults(&mut self) {
        let zstd_enabled_anywhere = self.zstd_wire_compression_enabled_anywhere();

        for store in &mut self.stores23 {
            let store_name = store.name.as_str();
            store.spec.visit_grpc_specs_mut(&mut |grpc| {9
                if !matches!3(grpc.store_type, StoreType::Cas) {
                    if grpc.experimental_remote_cache_compression == Some(true) {
                        warn!(
                            store = store_name,
                            instance_name = grpc.instance_name,
                            "'experimental_remote_cache_compression' is enabled on a non-CAS \
                             gRPC store, where it has no effect: REAPI zstd wire compression \
                             only applies to CAS blob transfers. Enable it on the CAS gRPC \
                             stores or a capabilities instance instead. See {}",
                            Self::ZSTD_COMPRESSION_DOCS_URL,
                        );
                    }
                    return;
                }

                if !zstd_enabled_anywhere {
                    return;
                }

                match grpc.experimental_remote_cache_compression {
                    None => grpc.experimental_remote_cache_compression = Some(true),
                    Some(false) => warn!(
                        store = store_name,
                        instance_name = grpc.instance_name,
                        "Zstd wire compression is enabled elsewhere, but this eligible CAS gRPC \
                         store explicitly disables it. Unless this upstream cannot use zstd, \
                         enable 'experimental_remote_cache_compression' for substantially faster \
                         transfers of compressible artifacts. See {}",
                        Self::ZSTD_COMPRESSION_DOCS_URL,
                    ),
                    Some(true) => {}
                }
            });
        }
    }

    fn check_store_conflict(services: &ServicesConfig) -> Result<(), Error> {
        if let Some(cas_config18) = &services.cas
            && let Some(ac_config) = &services.ac
        {
            // Create a hashmap from the CAS configuration for quick lookup
            let cas_store_map: HashMap<_, _> = cas_config
                .iter()
                .map18(|with_instance_name| {
                    (
                        &with_instance_name.instance_name,
                        &with_instance_name.cas_store,
                    )
                })
                .collect();

            for with_instance_name in ac_config18 {
                if let Some(cas_store) = cas_store_map.get(&with_instance_name.instance_name)
                    && cas_store == &&with_instance_name.ac_store
                {
                    return Err(make_err!(
                        Code::InvalidArgument,
                        "CAS and AC use the same store '{}' in the config",
                        cas_store
                    ));
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use tracing_test::traced_test;

    use super::*;

    fn grpc_compression_configs(config: &CasConfig) -> Vec<(bool, Option<bool>)> {
        let mut configs = Vec::new();
        for store in &config.stores4 {
            store.spec6.visit_grpc_specs6(&mut |grpc| {
                configs.push((
                    matches!2(grpc.store_type, StoreType::Cas),
                    grpc.experimental_remote_cache_compression,
                ));
            });
        }
        configs
    }

    #[test]
    fn capabilities_config_remote_cache_compression_deserializes_true() {
        let config: CapabilitiesConfig =
            serde_json5::from_str(r#"{"remote_cache_compression": true}"#).unwrap();

        assert!(config.remote_cache_compression);
    }

    #[test]
    fn capabilities_config_remote_cache_compression_defaults_false() {
        let config: CapabilitiesConfig = serde_json5::from_str("{}").unwrap();

        assert!(!config.remote_cache_compression);
    }

    #[test]
    fn omitted_grpc_compression_stays_disabled_without_zstd_intent() {
        let config = CasConfig::try_from_json5_str(
            r#"{
                stores: [{
                    name: "upstream",
                    grpc: {
                        endpoints: [{ address: "http://localhost:1234" }],
                        store_type: "cas",
                    },
                }],
                servers: [],
            }"#,
        )
        .unwrap();

        assert_eq!(grpc_compression_configs(&config), vec![(true, None)]);
    }

    #[test]
    fn explicit_grpc_zstd_intent_enables_other_eligible_grpc_stores() {
        let config = CasConfig::try_from_json5_str(
            r#"{
                stores: [
                    {
                        name: "enabled",
                        grpc: {
                            endpoints: [{ address: "http://localhost:1234" }],
                            store_type: "cas",
                            experimental_remote_cache_compression: true,
                        },
                    },
                    {
                        name: "inherited",
                        grpc: {
                            endpoints: [{ address: "http://localhost:5678" }],
                            store_type: "cas",
                        },
                    },
                ],
                servers: [],
            }"#,
        )
        .unwrap();

        assert_eq!(
            grpc_compression_configs(&config),
            vec![(true, Some(true)), (true, Some(true))]
        );
    }

    #[test]
    #[traced_test]
    fn ac_grpc_compression_warns_and_expresses_no_cas_intent() {
        let config = CasConfig::try_from_json5_str(
            r#"{
                stores: [
                    {
                        name: "action-cache",
                        grpc: {
                            instance_name: "ac",
                            endpoints: [{ address: "http://localhost:1234" }],
                            store_type: "ac",
                            experimental_remote_cache_compression: true,
                        },
                    },
                    {
                        name: "cas",
                        grpc: {
                            endpoints: [{ address: "http://localhost:5678" }],
                            store_type: "cas",
                        },
                    },
                ],
                servers: [],
            }"#,
        )
        .unwrap();

        // The AC-store setting is inert: it neither compresses AC RPCs nor
        // expresses process-wide zstd intent, so the CAS store stays unset.
        assert_eq!(
            grpc_compression_configs(&config),
            vec![(false, Some(true)), (true, None)]
        );
        assert!(logs_contain("store=\"action-cache\""));
        assert!(logs_contain("no effect"));
    }

    #[test]
    #[traced_test]
    fn capabilities_zstd_intent_defaults_nested_cas_grpc_and_warns_on_opt_out() {
        let config = CasConfig::try_from_json5_str(
            r#"{
                stores: [{
                    name: "nested-upstreams",
                    fast_slow: {
                        fast: {
                            grpc: {
                                instance_name: "auto",
                                endpoints: [{ address: "http://localhost:1234" }],
                                store_type: "cas",
                            },
                        },
                        slow: {
                            shard: {
                                stores: [
                                    {
                                        store: {
                                            grpc: {
                                                instance_name: "opt-out",
                                                endpoints: [{
                                                    address: "http://localhost:5678",
                                                }],
                                                store_type: "cas",
                                                experimental_remote_cache_compression: false,
                                            },
                                        },
                                        weight: 1,
                                    },
                                    {
                                        store: {
                                            grpc: {
                                                instance_name: "action-cache",
                                                endpoints: [{
                                                    address: "http://localhost:9012",
                                                }],
                                                store_type: "ac",
                                            },
                                        },
                                        weight: 1,
                                    },
                                ],
                            },
                        },
                    },
                }],
                servers: [{
                    listener: {
                        http: { socket_address: "127.0.0.1:50051" },
                    },
                    services: {
                        capabilities: [{
                            instance_name: "main",
                            remote_cache_compression: true,
                        }],
                    },
                }],
            }"#,
        )
        .unwrap();

        assert_eq!(
            grpc_compression_configs(&config),
            vec![(true, Some(true)), (true, Some(false)), (false, None),]
        );
        assert!(logs_contain("store=\"nested-upstreams\""));
        assert!(logs_contain("instance_name=\"opt-out\""));
        assert!(logs_contain(CasConfig::ZSTD_COMPRESSION_DOCS_URL));
    }
}

Coverage Report

Created: 2026-09-18 20:40