// 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 core::convert::Into;

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, InstanceName, 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, 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::{DigestHasher, DigestHasherFunc, make_ctx_for_hash_func};

use nativelink_util::spawn_blocking;

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, warn};

use crate::wire_compression::RemoteCacheCompressionInstances;

/// 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<MetricPublishKnownKindData, nativelink_metric::Error> {

        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<InstanceName, Store>,

    remote_cache_compression_instances: RemoteCacheCompressionInstances,

    chunking_instances: HashMap<InstanceName, ChunkingInstance>,

    chunking_metrics: ChunkingMetrics,

}

type GetTreeStream = Pin<Box<dyn Stream<Item = Result<GetTreeResponse, Status>> + Send + 'static>>;

/// 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<CasStoreConfig>],

        store_manager: &StoreManager,

        remote_cache_compression_instances: &RemoteCacheCompressionInstances,

    ) -> Result<Self, Error> {

        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(|| 
{0

                make_input_err!("'cas_store': '{}' does not exist", config.cas_store)

            })?;

            if let Some(
chunking_config13
) = &config.experimental_chunking {

                let avg_chunk_size_bytes = chunking_config

                    .validated_avg_chunk_size_bytes()

                    .err_tip(|| 
{0

                        format!(

                            "In 'experimental_chunking' of instance '{}'",

                            config.instance_name

                        )

                    })?;

                if store.downcast_ref::<GrpcStore>(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(|| 
{0

                    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(|| 
{0

                    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")
?0
;

                let max_chunk_count = usize::try_from(chunking_config.resolved_max_chunk_count())

                    .err_tip(|| "max_chunk_count did not fit in usize")
?0
;

                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,

            remote_cache_compression_instances: remote_cache_compression_instances.clone(),

            chunking_instances,

            chunking_metrics: ChunkingMetrics::default(),

        })

    }

    pub fn into_service(self) -> Server<Self> {

        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,

    ) -> Result<Response<FindMissingBlobsResponse>, Error> {

        let instance_name = &request.instance_name;

        let store = self

            .stores

            .get(instance_name)

            .err_tip(|| 
format!0
("'instance_name' not configured for '{instance_name}'"))
?0

            .clone();

        let mut requested_blobs = Vec::with_capacity(request.blob_digests.len());

        for digest in 
&request.blob_digests4
 {

            requested_blobs.
push6
(DigestInfo::try_from(digest.clone())
?1
.
into6
());

        }

        let sizes = store

            .has_many(&requested_blobs)

            .await

            .err_tip(|| "In find_missing_blobs")
?0
;

        let missing_blob_digests = sizes

            .into_iter()

            .zip(request.blob_digests)

            .
filter_map3
(|(maybe_size, digest)| maybe_size.map_or_else(|| Some(
digest2
), |_| None))

            .collect();

        Ok(Response::new(FindMissingBlobsResponse {

            missing_blob_digests,

        }))

    }

    async fn inner_batch_update_blobs(

        &self,

        request: BatchUpdateBlobsRequest,

    ) -> Result<Response<BatchUpdateBlobsResponse>, Error> {

        let instance_name = &request.instance_name;

        let store = self

            .stores

            .get(instance_name)

            .err_tip(|| 
format!0
("'instance_name' not configured for '{instance_name}'"))
?0

            .clone();

        // If we are a GrpcStore we shortcut here, as this is a special store.

        // Note: We don't know the digests here, so we try perform a very shallow

        // check to see if it's a grpc store.

        if let Some(
grpc_store0
) = store.downcast_ref::<GrpcStore>(None) {

            return grpc_store.batch_update_blobs(Request::new(request)).await;

        }

        let store_ref = &store;

        let remote_cache_compression_enabled = self

            .remote_cache_compression_instances

            .enabled_for(instance_name);

        let update_futures: FuturesUnordered<_> = request

            .requests

            .into_iter()

            .
map6
(|request| {

                async move {

                    let digest = request

                        .digest

                        .clone()

                        .err_tip(|| "Digest not found in request")
?0
;

                    let digest_info = DigestInfo::try_from(digest.clone())
?0
;

                    let size_bytes = usize::try_from(digest_info.size_bytes())

                        .err_tip(|| "Digest size_bytes was not convertible to usize")
?0
;

                    let 
store_data6
 = crate::wire_compression::decompress_batch_update(

                        request.data,

                        request.compressor,

                        size_bytes,

                        remote_cache_compression_enabled,

                    )

                    .await
?1
;

                    // Apply a per-blob deadline so one slow upload does not

                    // make the whole batch hit the client's overall deadline.

                    let result = match tokio::time::timeout(

                        BATCH_PER_BLOB_TIMEOUT,

                        store_ref.update_oneshot(digest_info, store_data),

                    )

                    .await

                    {

                        Ok(r) => r.err_tip(|| "Error writing to store"),

                        Err(_elapsed) => Err(make_err!(

                            Code::DeadlineExceeded,

                            "BatchUpdateBlobs per-blob timeout ({} s) elapsed for digest {}",

                            BATCH_PER_BLOB_TIMEOUT.as_secs(),

                            digest_info,

                        )),

                    };

                    Ok::<_, Error>(batch_update_blobs_response::Response {

                        digest: Some(digest),

                        status: Some(result.map_or_else(Into::into, |()| 
GrpcStatus::default5
())),

                    })

                }

            })

            .collect();

        let 
responses5
 = update_futures

            .try_collect::<Vec<batch_update_blobs_response::Response>>()

            .await
?1
;

        Ok(Response::new(BatchUpdateBlobsResponse { responses }))

    }

    async fn inner_batch_read_blobs(

        &self,

        request: BatchReadBlobsRequest,

    ) -> Result<Response<BatchReadBlobsResponse>, Error> {

        let instance_name = &request.instance_name;

        let store = self

            .stores

            .get(instance_name)

            .err_tip(|| 
format!0
("'instance_name' not configured for '{instance_name}'"))
?0

            .clone();

        // If we are a GrpcStore we shortcut here, as this is a special store.

        // Note: We don't know the digests here, so we try perform a very shallow

        // check to see if it's a grpc store.

        if let Some(
grpc_store0
) = store.downcast_ref::<GrpcStore>(None) {

            return grpc_store.batch_read_blobs(Request::new(request)).await;

        }

        let store_ref = &store;

        let remote_cache_compression_enabled = self

            .remote_cache_compression_instances

            .enabled_for(instance_name);

        let client_accepts_zstd = remote_cache_compression_enabled

            && 
request.acceptable_compressors.iter()3
.
any3
(|compressor_i32| {

                compressor::Value::try_from(*compressor_i32)

                    .is_ok_and(|compressor| 
compressor3
 == 
compressor::Value::Zstd3
)

            });

        let read_futures: FuturesUnordered<_> = request

            .digests

            .into_iter()

            .
map5
(|digest| {

                async move {

                    let digest_copy = DigestInfo::try_from(digest.clone())
?0
;

                    // TODO(palfrey) There is a security risk here of someone taking all the memory on the instance.

                    // Apply a per-blob deadline so one slow read does not

                    // make the whole batch hit the client's overall deadline.

                    let result = match tokio::time::timeout(

                        BATCH_PER_BLOB_TIMEOUT,

                        store_ref.get_part_unchunked(digest_copy, 0, None),

                    )

                    .await

                    {

                        Ok(r) => r.err_tip(|| "Error reading from store"),

                        Err(_elapsed) => Err(make_err!(

                            Code::DeadlineExceeded,

                            "BatchReadBlobs per-blob timeout ({} s) elapsed for digest {}",

                            BATCH_PER_BLOB_TIMEOUT.as_secs(),

                            digest_copy,

                        )),

                    };

                    let (status, data, response_compressor) = match result {

                        Ok(raw_data) => {

                            let (output_data, chosen_compressor) = if client_accepts_zstd {

                                let data_for_compression = raw_data.clone();

                                match spawn_blocking!("cas_encode_compressed_download", move || {

                                    crate::wire_compression::compress_for_batch_read(

                                        data_for_compression,

                                    )

                                })

                                .await

                                {

                                    Ok(compressed_result) => compressed_result,

                                    Err(e) => {

                                        warn!("Wire compression task failed for digest {:?}, falling back to identity: {}", digest, e);

                                        (raw_data, compressor::Value::Identity)

                                    }

                                }

                            } else {

                                (raw_data, compressor::Value::Identity)

                            };

                            (GrpcStatus::default(), output_data, chosen_compressor)

                        }

                        Err(mut e) => {

                            if e.code == Code::NotFound {

                                // Trim the error code. Not Found is quite common and we don't want to send a large

                                // error (debug) message for something that is common. We resize to just the last

                                // message as it will be the most relevant.

                                e.messages.resize_with(1, String::new);

                            }

                            (e.into(), Bytes::new(), compressor::Value::Identity)

                        }

                    };

                    Ok::<_, Error>(batch_read_blobs_response::Response {

                        status: Some(status),

                        digest: Some(digest),

                        compressor: response_compressor.into(),

                        data,

                    })

                }

            })

            .collect();

        let responses = read_futures

            .try_collect::<Vec<batch_read_blobs_response::Response>>()

            .await
?0
;

        Ok(Response::new(BatchReadBlobsResponse { responses }))

    }

    async fn inner_get_tree(

        &self,

        request: GetTreeRequest,

    ) -> Result<impl Stream<Item = Result<GetTreeResponse, Status>> + Send + use<>, Error> {

        let instance_name = &request.instance_name;

        let store = self

            .stores

            .get(instance_name)

            .err_tip(|| 
format!0
("'instance_name' not configured for '{instance_name}'"))
?0

            .clone();

        // If we are a GrpcStore we shortcut here, as this is a special store.

        // Note: We don't know the digests here, so we try perform a very shallow

        // check to see if it's a grpc store.

        if let Some(
grpc_store0
) = store.downcast_ref::<GrpcStore>(None) {

            let stream = grpc_store

                .get_tree(Request::new(request))

                .await?

                .into_inner();

            return Ok(stream.left_stream());

        }

        let root_digest: DigestInfo = request

            .root_digest

            .err_tip(|| "Expected root_digest to exist in GetTreeRequest")
?0

            .try_into()

            .err_tip(|| "In GetTreeRequest::root_digest")
?0
;

        let mut deque: VecDeque<DigestInfo> = VecDeque::new();

        let mut directories: Vec<Directory> = Vec::new();

        // `page_token` will return the `{hash_str}-{size_bytes}` of the current request's first directory digest.

        let page_token_digest = if request.page_token.is_empty() {

            root_digest

        } else {

            let mut page_token_parts = request.page_token.split('-');

            DigestInfo::try_new(

                page_token_parts

                    .next()

                    .err_tip(|| "Failed to parse `hash_str` in `page_token`")
?0
,

                page_token_parts

                    .next()

                    .err_tip(|| "Failed to parse `size_bytes` in `page_token`")
?0

                    .parse::<i64>()

                    .err_tip(|| "Failed to parse `size_bytes` as i64")
?0
,

            )

            .err_tip(|| "Failed to parse `page_token` as `Digest` in `GetTreeRequest`")
?0

        };

        let page_size = request.page_size;

        // If `page_size` is 0, paging is not necessary.

        let mut page_token_matched = page_size == 0;

        deque.push_back(root_digest);

        while !deque.is_empty() {

            let digest: DigestInfo = deque.pop_front().err_tip(|| "In VecDeque::pop_front")
?0
;

            let directory = get_and_decode_digest::<Directory>(&store, digest.into())

                .await

                .err_tip(|| "Converting digest to Directory")
?0
;

            if digest == page_token_digest {

                page_token_matched = true;

            }

            for directory in 
&directory.directories26
 {

                let digest: DigestInfo = directory

                    .digest

                    .clone()

                    .err_tip(|| "Expected Digest to exist in Directory::directories::digest")
?0

                    .try_into()

                    .err_tip(|| "In Directory::file::digest")
?0
;

                deque.push_back(digest);

            }

            let page_size_usize = usize::try_from(page_size).unwrap_or(usize::MAX);

            if page_token_matched {

                directories.push(directory);

                if directories.len() == page_size_usize {

                    break;

                }

            }

        }

        // `next_page_token` will return the `{hash_str}:{size_bytes}` of the next request's first directory digest.

        // It will be an empty string when it reached the end of the directory tree.

        let next_page_token: String = deque

            .front()

            .map_or_else(String::new, |value| 
format!3
("{value}"));

        Ok(futures::stream::once(async {

            Ok(GetTreeResponse {

                directories,

                next_page_token,

            })

        })

        .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!0
("'instance_name' not configured for '{instance_name}'"))
?0

            .clone();

        let 
chunking_instance14
 = self

            .chunking_instances

            .get(instance_name)

            .ok_or_else(|| 
{2

                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::<GrpcStore>(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<DigestHasherFunc> {

        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<DigestHasherFunc, Error> {

        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 hashers1
 = CANDIDATES.
map1
(|func| func.hasher());

            loop {

                let data = rx

                    .recv()

                    .await

                    .err_tip(|| "In infer_blob_hasher_func::recv")
?0
;

                if data.is_empty() {

                    break; // EOF.

                }

                for hasher in 
&mut hashers1
 {

                    hasher.update(&data);

                }

            }

            Ok::<_, Error>(
hashers1
.
map1
(|mut hasher| hasher.finalize_digest()))

        };

        let (read_res, hash_res) = futures::join!(read_fut, hash_fut);

        let computed_digests = read_res.merge(hash_res)
?0
;

        CANDIDATES

            .iter()

            .zip(computed_digests)

            .
find1
(|(_, computed)| *computed == blob_digest)

            .map(|(func, _)| *func)

            .ok_or_else(|| 
{0

                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<Vec<String>, Error> {

        let mut digest_infos = Vec::with_capacity(chunk_digests.len());

        for digest in 
chunk_digests7
 {

            digest_infos

                .push(DigestInfo::try_from(digest.clone()).err_tip(|| "Invalid chunk digest")
?0
);

        }

        let 
chunk_keys7
: 
Vec<_>7
 = 
digest_infos.iter()7
.
map7
(|digest| (*digest).into()).
collect7
();

        let sizes = store

            .has_many(&chunk_keys)

            .await

            .err_tip(|| "In missing_chunks")
?0
;

        Ok(sizes

            .iter()

            .zip(&digest_infos)

            .
filter7
(|(maybe_size, _)| maybe_size.is_none())

            .map(|(_, digest)| 
digest2
.
to_string2
())

            .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<SplitBlobResponse> {

        let 
layout_bytes3
 = chunking_instance

            .index_store

            .get_part_unchunked(blob_digest, 0, Some(chunking_instance.max_layout_size()))

            .await

            .ok()
?5
;

        let layout = SplitBlobResponse::decode(layout_bytes).ok()
?0
;

        // 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_digests3
 {

            total_size = total_size.checked_add(u64::try_from(digest.size_bytes).ok()
?0
)
?0
;

        }

        (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<Response<SplitBlobResponse>, 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_store0
) = self.grpc_store_for_instance(&request.instance_name) {

            return grpc_store.split_blob(Request::new(request)).await;

        }

        let (
store8
, 
chunking_instance8
) = self.chunking_instance(&request.instance_name)
?1
;

        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")
?0

            .try_into()

            .err_tip(|| "In SplitBlobRequest::blob_digest")
?0
;

        // 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")
?0
.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(
layout3
) = maybe_layout

            && matches!(

                Self::missing_chunks(&store, &layout.chunk_digests).await,

                Ok(
missing2
) if missing.is_empty(
)2

            )

        {

            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_response4
 = self

            .chunk_blob_on_demand(

                &store,

                &chunking_instance,

                blob_digest,

                request.digest_function,

            )

            .await
?1
;

        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<SplitBlobResponse, Error> {

        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
?0
,

        };

        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!0
("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<Digest> = pin!(cdc.as_stream())

                .
map5
(|chunk_result| async {

                    let chunk = chunk_result

                        .map_err(|e| 
make_err!0
(
Code::Internal0
, "Failed to chunk blob: {e:?}"))

                        .err_tip(|| "In chunk_blob_on_demand")
?0
;

                    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")
?0

                        .is_none()

                    {

                        store

                            .update_oneshot(chunk_digest, chunk.data.into())

                            .await

                            .err_tip(|| 
{0

                                format!(

                                    "Failed to store chunk {chunk_digest} in chunk_blob_on_demand"

                                )

                            })?;

                    }

                    Ok::<Digest, Error>(chunk_digest.into())

                })

                .buffered(CHUNK_CONCURRENCY)

                .try_collect()

                .await
?0
;

            Ok::<Vec<Digest>, 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")
?0
;

        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
?0
;

        Ok(split_response)

    }

    async fn inner_splice_blob(

        &self,

        request: SpliceBlobRequest,

    ) -> Result<Response<SpliceBlobResponse>, 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_store0
) = self.grpc_store_for_instance(&request.instance_name) {

            return grpc_store.splice_blob(Request::new(request)).await;

        }

        let (
store6
, 
chunking_instance6
) = self.chunking_instance(&request.instance_name)
?1
;

        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")
?0

            .try_into()

            .err_tip(|| "In SpliceBlobRequest::blob_digest")
?0
;

        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_digests5
 {

            let digest_info = DigestInfo::try_from(digest.clone())

                .err_tip(|| "In SpliceBlobRequest::chunk_digests")
?0
;

            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")
?0
;

        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.