/build/source/nativelink-service/src/fetch_server.rs

Source
// Copyright 2025 Trace Machina, Inc. 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 std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64;
use bytes::Bytes;
use nativelink_config::cas_server::{FetchConfig, FetchDownloaderConfig, WithInstanceName};
use nativelink_error::{Error, ResultExt, make_err, make_input_err};
use nativelink_proto::build::bazel::remote::asset::v1::fetch_server::{
    Fetch, FetchServer as Server,
};
use nativelink_proto::build::bazel::remote::asset::v1::{
    FetchBlobRequest, FetchBlobResponse, FetchDirectoryRequest, FetchDirectoryResponse,
};
use nativelink_proto::google::rpc::Status as GoogleStatus;
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, default_digest_hasher_func, digest_hasher_func_from_context,
    make_ctx_for_hash_func,
};
use nativelink_util::store_trait::{Store, StoreLike, UploadSizeInfo};
use opentelemetry::context::FutureExt;
use prost::Message;
use reqwest::{Client, NoProxy, Proxy};
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
use tokio::sync::Semaphore;
use tonic::{Code, Request, Response, Status};
use tracing::{Instrument, Level, error_span, info, instrument};

use crate::remote_asset_proto::{RemoteAssetArtifact, RemoteAssetQuery};

#[derive(Debug, Clone)]
pub struct FetchStoreInfo {
    store: Store,
    downloader: Option<Arc<Downloader>>,
}

#[derive(Debug)]
struct ExpectedChecksum {
    function: DigestHasherFunc,
    hash: [u8; 32],
}

#[derive(Debug)]
struct Downloader {
    client: Client,
    proxy_client: Option<Client>,
    require_checksum: bool,
    allow_unknown_qualifiers: bool,
    cas: Store,
    max_size: Option<u64>,
    timeout: Duration,
    slots: Semaphore,
}

impl Downloader {
    fn new(config: &FetchDownloaderConfig, stores: &StoreManager) -> Result<Self, Error> {
        if config.http_proxy.is_some() && config.http_proxy_secret_file9.is_some9() {
            return Err(make_input_err!("Set only one fetch proxy URL source"));
        }
        if config.max_concurrent_downloads > Semaphore::MAX_PERMITS {
            return Err(make_input_err!(
                "Fetch max_concurrent_downloads is too large"
            ));
        }
        let proxy_url41 = match &config.http_proxy_secret_file {
            Some(path) => Some(
                std::fs::read_to_string(path)
                    .map_err(|_| make_input_err!1("Could not read fetch proxy secret file"))?1,
            ),
            None => config.http_proxy.clone(),
        };
        // Match the server's TLS provider, including memory-only store configurations.
        drop(rustls::crypto::ring::default_provider().install_default());
        let builder = Client::builder()
            // Redirect destinations must not receive credentials from the source URL.
            .referer(false)
            .no_proxy()
            .redirect(reqwest::redirect::Policy::limited(10));
        let proxy_client40 = if let Some(url9) = proxy_url {
            let url = url.trim();
            let parsed =
                reqwest::Url::parse(url).map_err(|_| make_input_err!1("Invalid fetch proxy URL"))?1;
            if !matches!(parsed.scheme(), "http" | "https"0) || parsed.host_str().is_none() {
                return Err(make_input_err!("Fetch proxy must be an HTTP(S) URL"));
            }
            let bypass = NoProxy::from_string(&config.no_proxy.join(","));
            let proxy = Proxy::all(url)
                .map_err(|_| make_input_err!0("Invalid fetch proxy URL"))?0
                .no_proxy(bypass);
            Some(
                Client::builder()
                    .referer(false)
                    .redirect(reqwest::redirect::Policy::limited(10))
                    .no_proxy()
                    .proxy(proxy)
                    .build()
                    .map_err(|_| make_input_err!0("Could not build fetch proxy client"))?0,
            )
        } else {
            None
        };
        let cas = stores
            .get_store(&config.cas_store)
            .ok_or_else(|| make_input_err!0("Fetch downloader CAS store does not exist"))?0;
        Ok(Self {
            client: builder
                .build()
                .map_err(|_| make_input_err!0("Could not build fetch HTTP client"))?0,
            cas,
            proxy_client,
            require_checksum: config.require_checksum,
            allow_unknown_qualifiers: config.allow_unknown_qualifiers,
            max_size: (config.max_size_bytes != 0).then_some(config.max_size_bytes),
            timeout: Duration::from_secs(if config.timeout_s == 0 {
                300
            } else {
                config.timeout_s
            }),
            slots: Semaphore::new(if config.max_concurrent_downloads == 0 {
                4
            } else {
                config.max_concurrent_downloads
            }),
        })
    }

    async fn download(
        &self,
        uri: &str,
        checksum: Option<&[ExpectedChecksum]>,
    ) -> Result<DigestInfo, Error> {
        let default = Box::pin(self.download_with_client(&self.client, uri, checksum));
        let Some(proxy9) = &self.proxy_client else {
            return default.await;
        };
        // A failed path does not cancel an alternative that can still succeed.
        let attempts = vec![
            default,
            Box::pin(self.download_with_client(proxy, uri, checksum)),
        ];
        futures::future::select_ok(attempts)
            .await
            .map(|(digest, _)| digest)
    }

    async fn download_with_client(
        &self,
        client: &Client,
        uri: &str,
        checksum: Option<&[ExpectedChecksum]>,
    ) -> Result<DigestInfo, Error> {
        let url = reqwest::Url::parse(uri).map_err(|_| make_input_err!0("Invalid fetch URI"))?0;
        if !matches!(url.scheme(), "http" | "https"0) {
            return Err(make_input_err!("Only HTTP(S) fetch URIs are supported"));
        }
        let mut response56 = client
            .get(url)
            .send()
            .await
            .map_err(|_| make_err!0(Code::Unavailable0, "Origin request failed"))?0
            .error_for_status()
            .map_err(|_| make_err!2(Code::Unavailable2, "Origin returned an HTTP error"))?2;
        if response.status() != reqwest::StatusCode::OK {
            return Err(make_err!(
                Code::Unavailable,
                "Origin did not return a complete object"
            ));
        }
        if self
            .max_size
            .is_some_and(|limit| response54.content_length54().is_some_and54(|size| size3 > limit3))
        {
            return Err(make_err!(
                Code::ResourceExhausted,
                "Fetch object exceeds max_size_bytes"
            ));
        }
        // Spool before publishing: no unverified object enters CAS, even without a checksum.
        // The anonymous file is removed on close, including timeout and cancellation.
        let file = nativelink_util::spawn_blocking!("fetch_spool_create", tempfile::tempfile)
            .await
            .map_err(|_| make_err!0(Code::Internal0, "Could not create fetch spool"))?0?0;
        let mut file = tokio::fs::File::from_std(file);
        let digest_function = digest_hasher_func_from_context();
        let mut hasher = digest_function.hasher();
        let mut checksum_hashers = Vec::new();
        for expected33 in checksum.into_iter().flatten() {
            if expected.function != digest_function
                && !checksum_hashers
                    .iter()
                    .any(|(function, _)| *function2 == expected.function2)
            {
                checksum_hashers.push((expected.function, expected.function.hasher()));
            }
        }
        let mut size = 0_u64;
        while let Some(chunk1.23k) = response
            .chunk()
            .await
            .map_err(|_| make_err!0(Code::Unavailable0, "Origin body transfer failed"))?0
        {
            size = size
                .checked_add(chunk.len() as u64)
                .filter(|n| self.max_size.is_none_or(|limit| *n53 <= limit53))
                .ok_or_else(|| {1
                    make_err!(
                        Code::ResourceExhausted,
                        "Fetch object exceeds max_size_bytes"
                    )
                })?;
            hasher.update(&chunk);
            for (_, hasher8) in &mut checksum_hashers {
                hasher.update(&chunk);
            }
            file.write_all(&chunk).await?0;
        }
        let digest = hasher.finalize_digest();
        if let Some(alternatives24) = checksum {
            let checksum_digests: Vec<_> = checksum_hashers
                .into_iter()
                .map(|(function, mut hasher)| (function8, hasher8.finalize_digest8()))
                .collect();
            let matches24 = alternatives.iter()24.any24(|expected| {
                let actual = if expected.function == digest_function {
                    Some(&digest)
                } else {
                    checksum_digests
                        .iter()
                        .find(|(function, _)| *function == expected.function)
                        .map(|(_, digest)| digest)
                };
                actual.is_some_and(|actual| {
                    actual.packed_hash().as_slice() == expected.hash.as_slice()
                })
            });
            if !matches {
                return Err(make_err!(Code::DataLoss, "Fetch checksum mismatch"));
            }
        }
        file.rewind().await?0;
        let (mut tx, rx) = make_buf_channel_pair();
        let replay = async {
            let mut buffer = vec![0; 64 * 1024];
            loop {
                let count4.18k = file.read(&mut buffer).await?0;
                if count == 0 {
                    break;
                }
                tx.send(Bytes::copy_from_slice(&buffer[..count])).await?0;
            }
            tx.send_eof()
        };
        futures::try_join!(
            replay,
            self.cas.update(digest, rx, UploadSizeInfo::ExactSize(size))
        )?;
        Ok(digest)
    }
}

#[derive(Debug, Clone)]
pub struct FetchServer {
    stores: HashMap<String, FetchStoreInfo>,
}

impl FetchServer {
    pub fn new(
        configs: &[WithInstanceName<FetchConfig>],
        store_manager: &StoreManager,
    ) -> Result<Self, Error> {
        let mut stores = HashMap::with_capacity(configs.len());
        for config in configs47 {
            let store = store_manager
                .get_store(&config.fetch_store)
                .ok_or_else(|| {0
                    make_input_err!("'fetch_store': '{}' does not exist", config.fetch_store)
                })?;
            let downloader45 = config
                .downloader
                .as_ref()
                .map(|spec| Downloader::new43(spec43, store_manager43).map43(Arc::new))
                .transpose()?3;
            stores.insert(
                config.instance_name.clone(),
                FetchStoreInfo { store, downloader },
            );
        }
        Ok(Self { stores })
    }

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

    async fn inner_fetch_blob(
        &self,
        request: FetchBlobRequest,
    ) -> Result<Response<FetchBlobResponse>, Error> {
        let instance_name = &request.instance_name;
        let store_info = self.stores.get(instance_name).ok_or_else(|| {0
            make_input_err!("'instance_name' not configured for '{instance_name}'")
        })?;

        if request.uris.is_empty() {
            return Err(Error::new(
                Code::InvalidArgument,
                "No uris in fetch request".to_owned(),
            ));
        }
        let download_options83 = if let Some(downloader94) = &store_info.downloader {
            let checksum78 = Self::checksum(&request, downloader)?16;
            let timeout = match request.timeout {
                Some(value) => Duration::try_from(value)
                    .map_err(|_| make_input_err!0("Invalid fetch timeout"))?0
                    .min(downloader.timeout),
                None => downloader.timeout,
            };
            Some((downloader, checksum, timeout))
        } else {
            None
        };
        for uri in &request.uris83 {
            let asset_request = RemoteAssetQuery::new(uri.clone(), request.qualifiers.clone());
            let asset_digest = asset_request.digest();
            let asset_response_possible = store_info
                .store
                .get_part_unchunked(asset_digest, 0, None)
                .await;

            info!(
                digest = format!("{}", asset_digest),
                "Looked up fetch asset"
            );

            if let Ok(asset_response_raw34) = asset_response_possible {
                let asset_response33 = RemoteAssetArtifact::decode(asset_response_raw)
                    .err_tip(|| "Failed to decode stored RemoteAssetArtifact")?1;
                if let Some(downloader30) = &store_info.downloader {
                    // Legacy index keys concatenate fields without boundaries. A matching
                    // key alone does not prove that the record belongs to this request.
                    if asset_response.uri != asset_request.uri
                        || asset_response.qualifiers != asset_request.qualifiers
                    {
                        continue;
                    }
                    let now: prost_types::Timestamp = SystemTime::now().into();
                    let unexpired = asset_response.expire_at.as_ref().is_none_or(|expiry| {0
                        (expiry.seconds, expiry.nanos) > (now.seconds, now.nanos)
                    });
                    let stored_function =
                        DigestHasherFunc::try_from(asset_response.digest_function)?0;
                    let same_function = stored_function == digest_hasher_func_from_context();
                    let fresh_enough =
                        request
                            .oldest_content_accepted
                            .as_ref()
                            .is_none_or(|oldest| {7
                                asset_response.fetched_at.as_ref().is_some_and(|fetched| {6
                                    (fetched.seconds, fetched.nanos)
                                        >= (oldest.seconds, oldest.nanos)
                                })
                            });
                    if !unexpired || !same_function || !fresh_enough {
                        continue;
                    }
                    let Some(blob) = asset_response.blob_digest.as_ref() else {
                        continue;
                    };
                    if downloader
                        .cas
                        .has(DigestInfo::try_from(blob)?0)
                        .await?0
                        .is_none()
                    {
                        continue;
                    }
                }
                return Ok(Response::new(FetchBlobResponse {
                    status: Some(GoogleStatus {
                        code: Code::Ok.into(),
                        message: "Fetch object found".to_owned(),
                        details: vec![],
                    }),
                    uri: asset_response.uri,
                    qualifiers: asset_response.qualifiers,
                    expires_at: asset_response.expire_at,
                    blob_digest: asset_response.blob_digest,
                    digest_function: asset_response.digest_function,
                }));
            }
        }
        let Some((downloader56, checksum56, timeout56)) = download_options else {
            let mut response =
                Self::miss_response(&request, make_err!(Code::NotFound, "No item found"));
            response.digest_function = default_digest_hasher_func().proto_digest_func().into();
            return Ok(Response::new(response));
        };
        let retrieve = async {
            let _slot = downloader
                .slots
                .acquire()
                .await
                .map_err(|_| make_err!0(Code::Unavailable0, "Fetch downloader closed"))?0;
            let mut last_error = make_err!(Code::NotFound, "No downloadable URI found");
            for uri in &request.uris56 {
                let fetched_at = SystemTime::now().into();
                match downloader.download(uri, checksum.as_deref()).await {
                    Ok(digest) => {
                        let mut artifact = RemoteAssetArtifact::new(
                            uri.clone(),
                            request.qualifiers.clone(),
                            digest.into(),
                            None,
                            digest_hasher_func_from_context().proto_digest_func().into(),
                        );
                        artifact.fetched_at = Some(fetched_at);
                        store_info
                            .store
                            .update_oneshot(artifact.digest(), artifact.as_bytes())
                            .await?0;
                        return Ok(Response::new(FetchBlobResponse {
                            status: Some(GoogleStatus {
                                code: 0,
                                message: "Fetch object created".into(),
                                details: vec![],
                            }),
                            uri: artifact.uri,
                            qualifiers: artifact.qualifiers,
                            expires_at: artifact.expire_at,
                            blob_digest: artifact.blob_digest,
                            digest_function: artifact.digest_function,
                        }));
                    }
                    Err(error) => last_error = error,
                }
            }
            Ok(Response::new(Self::miss_response(&request, last_error)))
        };
        tokio::time::timeout(timeout, retrieve)
            .await
            .unwrap_or_else(|_| {1
                Ok(Response::new(Self::miss_response(
                    &request,
                    make_err!(Code::DeadlineExceeded, "Fetch origin timeout exceeded"),
                )))
            })
    }

    fn checksum(
        request: &FetchBlobRequest,
        downloader: &Downloader,
    ) -> Result<Option<Vec<ExpectedChecksum>>, Error> {
        let mut checksum = None;
        for qualifier60 in &request.qualifiers {
            if qualifier.name.starts_with("http_header:")
                || qualifier.name.starts_with("http_header_url:")
            {
                return Err(make_input_err!(
                    "HTTP header qualifiers are not supported by the fetch downloader"
                ));
            }
            if qualifier.name != "checksum.sri" {
                if qualifier.name != "bazel.canonical_id" && !downloader.allow_unknown_qualifiers {
                    return Err(make_input_err!("Unsupported fetch qualifier"));
                }
                continue;
            }
            if checksum.is_some() {
                return Err(make_input_err!("Duplicate checksum qualifier"));
            }
            let mut alternatives = Vec::new();
            for token in qualifier.value.split_whitespace()43 {
                let (algorithm58, encoded58) = token
                    .split_once('-')
                    .ok_or_else(|| make_input_err!1("Invalid checksum.sri format"))?1;
                if algorithm.is_empty() || encoded.is_empty() {
                    return Err(make_input_err!("Invalid checksum.sri format"));
                }
                let hash56 = BASE64
                    .decode(encoded)
                    .map_err(|_| make_input_err!2("Invalid checksum.sri encoding"))?2;
                // Unsupported algorithms may coexist with supported alternatives.
                let Ok(function54) = DigestHasherFunc::try_from(algorithm) else {
                    continue;
                };
                let hash53 = hash
                    .try_into()
                    .map_err(|_| make_input_err!1("Invalid checksum.sri length"))?1;
                alternatives.push(ExpectedChecksum { function, hash });
            }
            if alternatives.is_empty() {
                return Err(make_input_err!("No supported checksum.sri alternative"));
            }
            checksum = Some(alternatives);
        }
        if downloader.require_checksum && checksum2.is_none2() {
            return Err(make_input_err!("Fetch downloader requires checksum.sri"));
        }
        Ok(checksum)
    }

    fn miss_response(request: &FetchBlobRequest, error: Error) -> FetchBlobResponse {
        FetchBlobResponse {
            status: Some(error.into()),
            uri: request.uris.first().cloned().unwrap_or_default(),
            qualifiers: vec![],
            expires_at: None,
            blob_digest: None,
            digest_function: digest_hasher_func_from_context().proto_digest_func().into(),
        }
    }
}

#[tonic::async_trait]
impl Fetch for FetchServer {
    #[allow(clippy::blocks_in_conditions)]
    #[instrument(
        err(level = Level::WARN),
        skip_all,
        fields(
            uri_count = grpc_request.get_ref().uris.len(),
            qualifier_count = grpc_request.get_ref().qualifiers.len(),
            digest_function = grpc_request.get_ref().digest_function,
        )
    )]
    async fn fetch_blob(
        &self,
        grpc_request: Request<FetchBlobRequest>,
    ) -> Result<Response<FetchBlobResponse>, Status> {
        let request = grpc_request.into_inner();
        let digest_function = request.digest_function;
        self.inner_fetch_blob(request)
            .instrument(error_span!("fetch_server_fetch_blob"))
            .with_context(
                make_ctx_for_hash_func(digest_function).err_tip(|| "In FetchServer::fetch_blob")?,
            )
            .await
            .err_tip(|| "Failed on fetch_blob() command")
            .map_err(Into::into)
    }

    #[allow(clippy::blocks_in_conditions)]
    #[instrument(
        err(level = Level::WARN),
        skip_all,
        fields(
            uri_count = _grpc_request.get_ref().uris.len(),
            qualifier_count = _grpc_request.get_ref().qualifiers.len(),
            digest_function = _grpc_request.get_ref().digest_function,
        )
    )]
    async fn fetch_directory(
        &self,
        _grpc_request: Request<FetchDirectoryRequest>,
    ) -> Result<Response<FetchDirectoryResponse>, Status> {
        Err(Status::unimplemented("FetchDirectory not implemented"))
    }
}

Coverage Report

Created: 2026-11-13 23:29

Line	Count	Source
1		// Copyright 2025 Trace Machina, Inc. All rights reserved.
2		//
3		// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4		// you may not use this file except in compliance with the License.
5		// You may obtain a copy of the License at
6		//
7		// See LICENSE file for details
8		//
9		// Unless required by applicable law or agreed to in writing, software
10		// distributed under the License is distributed on an "AS IS" BASIS,
11		// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12		// See the License for the specific language governing permissions and
13		// limitations under the License.
14
15		use core::convert::Into;
16		use std::collections::HashMap;
17		use std::sync::Arc;
18		use std::time::{Duration, SystemTime};
19
20		use base64::Engine as _;
21		use base64::engine::general_purpose::STANDARD as BASE64;
22		use bytes::Bytes;
23		use nativelink_config::cas_server::{FetchConfig, FetchDownloaderConfig, WithInstanceName};
24		use nativelink_error::{Error, ResultExt, make_err, make_input_err};
25		use nativelink_proto::build::bazel::remote::asset::v1::fetch_server::{
26		Fetch, FetchServer as Server,
27		};
28		use nativelink_proto::build::bazel::remote::asset::v1::{
29		FetchBlobRequest, FetchBlobResponse, FetchDirectoryRequest, FetchDirectoryResponse,
30		};
31		use nativelink_proto::google::rpc::Status as GoogleStatus;
32		use nativelink_store::store_manager::StoreManager;
33		use nativelink_util::buf_channel::make_buf_channel_pair;
34		use nativelink_util::common::DigestInfo;
35		use nativelink_util::digest_hasher::{
36		DigestHasher, DigestHasherFunc, default_digest_hasher_func, digest_hasher_func_from_context,
37		make_ctx_for_hash_func,
38		};
39		use nativelink_util::store_trait::{Store, StoreLike, UploadSizeInfo};
40		use opentelemetry::context::FutureExt;
41		use prost::Message;
42		use reqwest::{Client, NoProxy, Proxy};
43		use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
44		use tokio::sync::Semaphore;
45		use tonic::{Code, Request, Response, Status};
46		use tracing::{Instrument, Level, error_span, info, instrument};
47
48		use crate::remote_asset_proto::{RemoteAssetArtifact, RemoteAssetQuery};
49
50		#[derive(Debug, Clone)]
51		pub struct FetchStoreInfo {
52		store: Store,
53		downloader: Option<Arc<Downloader>>,
54		}
55
56		#[derive(Debug)]
57		struct ExpectedChecksum {
58		function: DigestHasherFunc,
59		hash: [u8; 32],
60		}
61
62		#[derive(Debug)]
63		struct Downloader {
64		client: Client,
65		proxy_client: Option<Client>,
66		require_checksum: bool,
67		allow_unknown_qualifiers: bool,
68		cas: Store,
69		max_size: Option<u64>,
70		timeout: Duration,
71		slots: Semaphore,
72		}
73
74		impl Downloader {
75	43	fn new(config: &FetchDownloaderConfig, stores: &StoreManager) -> Result<Self, Error> {
76	43	if config.http_proxy.is_some() && config.http_proxy_secret_file9 . is_some9 () {
77	1	return Err(make_input_err!("Set only one fetch proxy URL source"));
78	42	}
79	42	if config.max_concurrent_downloads > Semaphore::MAX_PERMITS {
80	0	return Err(make_input_err!(
81	0	"Fetch max_concurrent_downloads is too large"
82	0	));
83	42	}
84	42	let proxy_url41 = match &config.http_proxy_secret_file {
85	2	Some(path) => Some(
86	2	std::fs::read_to_string(path)
87	2	.map_err(\|_\| make_input_err!1 ("Could not read fetch proxy secret file")) ?1 ,
88		),
89	40	None => config.http_proxy.clone(),
90		};
91		// Match the server's TLS provider, including memory-only store configurations.
92	41	drop(rustls::crypto::ring::default_provider().install_default());
93	41	let builder = Client::builder()
94		// Redirect destinations must not receive credentials from the source URL.
95	41	.referer(false)
96	41	.no_proxy()
97	41	.redirect(reqwest::redirect::Policy::limited(10));
98	41	let proxy_client40 = if let Some( url9 ) = proxy_url {
99	9	let url = url.trim();
100	8	let parsed =
101	9	reqwest::Url::parse(url).map_err(\|_\| make_input_err!1 ("Invalid fetch proxy URL")) ?1 ;
102	8	if !matches!(parsed.scheme(), "http" \| "https"0 ) \|\| parsed.host_str().is_none() {
103	0	return Err(make_input_err!("Fetch proxy must be an HTTP(S) URL"));
104	8	}
105	8	let bypass = NoProxy::from_string(&config.no_proxy.join(","));
106	8	let proxy = Proxy::all(url)
107	8	.map_err(\|_\| make_input_err!0 ("Invalid fetch proxy URL")) ?0
108	8	.no_proxy(bypass);
109		Some(
110	8	Client::builder()
111	8	.referer(false)
112	8	.redirect(reqwest::redirect::Policy::limited(10))
113	8	.no_proxy()
114	8	.proxy(proxy)
115	8	.build()
116	8	.map_err(\|_\| make_input_err!0 ("Could not build fetch proxy client")) ?0 ,
117		)
118		} else {
119	32	None
120		};
121	40	let cas = stores
122	40	.get_store(&config.cas_store)
123	40	.ok_or_else(\|\| make_input_err!0 ("Fetch downloader CAS store does not exist")) ?0 ;
124		Ok(Self {
125	40	client: builder
126	40	.build()
127	40	.map_err(\|_\| make_input_err!0 ("Could not build fetch HTTP client")) ?0 ,
128	40	cas,
129	40	proxy_client,
130	40	require_checksum: config.require_checksum,
131	40	allow_unknown_qualifiers: config.allow_unknown_qualifiers,
132	40	max_size: (config.max_size_bytes != 0).then_some(config.max_size_bytes),
133	40	timeout: Duration::from_secs(if config.timeout_s == 0 {
134	0	300
135		} else {
136	40	config.timeout_s
137		}),
138	40	slots: Semaphore::new(if config.max_concurrent_downloads == 0 {
139	1	4
140		} else {
141	39	config.max_concurrent_downloads
142		}),
143		})
144	43	}
145
146	57	async fn download(
147	57	&self,
148	57	uri: &str,
149	57	checksum: Option<&[ExpectedChecksum]>,
150	57	) -> Result<DigestInfo, Error> {
151	57	let default = Box::pin(self.download_with_client(&self.client, uri, checksum));
152	57	let Some( proxy9 ) = &self.proxy_client else {
153	48	return default.await;
154		};
155		// A failed path does not cancel an alternative that can still succeed.
156	9	let attempts = vec![
157	9	default,
158	9	Box::pin(self.download_with_client(proxy, uri, checksum)),
159		];
160	9	futures::future::select_ok(attempts)
161	9	.await
162	9	.map(\|(digest, _)\| digest)
163	56	}
164
165	66	async fn download_with_client(
166	66	&self,
167	66	client: &Client,
168	66	uri: &str,
169	66	checksum: Option<&[ExpectedChecksum]>,
170	66	) -> Result<DigestInfo, Error> {
171	66	let url = reqwest::Url::parse(uri).map_err(\|_\| make_input_err!0 ("Invalid fetch URI")) ?0 ;
172	66	if !matches!(url.scheme(), "http" \| "https"0 ) {
173	0	return Err(make_input_err!("Only HTTP(S) fetch URIs are supported"));
174	66	}
175	66	let mut response56 = client
176	66	.get(url)
177	66	.send()
178	66	.await
179	58	.map_err(\|_\| make_err!0 ( Code::Unavailable0 , "Origin request failed")) ?0
180	58	.error_for_status()
181	58	.map_err(\|_\| make_err!2 ( Code::Unavailable2 , "Origin returned an HTTP error")) ?2 ;
182	56	if response.status() != reqwest::StatusCode::OK {
183	1	return Err(make_err!(
184	1	Code::Unavailable,
185	1	"Origin did not return a complete object"
186	1	));
187	55	}
188	55	if self
189	55	.max_size
190	55	.is_some_and(\|limit\| response54 . content_length54 (). is_some_and54 (\|size\| size3 > limit3 ))
191		{
192	0	return Err(make_err!(
193	0	Code::ResourceExhausted,
194	0	"Fetch object exceeds max_size_bytes"
195	0	));
196	55	}
197		// Spool before publishing: no unverified object enters CAS, even without a checksum.
198		// The anonymous file is removed on close, including timeout and cancellation.
199	55	let file = nativelink_util::spawn_blocking!("fetch_spool_create", tempfile::tempfile)
200	55	.await
201	55	.map_err(\|_\| make_err!0 ( Code::Internal0 , "Could not create fetch spool")) ?0 ?0 ;
202	55	let mut file = tokio::fs::File::from_std(file);
203	55	let digest_function = digest_hasher_func_from_context();
204	55	let mut hasher = digest_function.hasher();
205	55	let mut checksum_hashers = Vec::new();
206	55	for expected33 in checksum.into_iter().flatten() {
207	33	if expected.function != digest_function
208	10	&& !checksum_hashers
209	10	.iter()
210	10	.any(\|(function, _)\| *function2 == expected.function2 )
211	8	{
212	8	checksum_hashers.push((expected.function, expected.function.hasher()));
213	25	}
214		}
215	55	let mut size = 0_u64;
216	1.28k	while let Some( chunk1.23k ) = response
217	1.28k	.chunk()
218	1.28k	.await
219	1.28k	.map_err(\|_\| make_err!0 ( Code::Unavailable0 , "Origin body transfer failed")) ?0
220		{
221	1.23k	size = size
222	1.23k	.checked_add(chunk.len() as u64)
223	1.23k	.filter(\|n\| self.max_size.is_none_or(\|limit\| *n53 <= limit53 ))
224	1.23k	.ok_or_else(\|\| {1
225	1	make_err!(
226	1	Code::ResourceExhausted,
227		"Fetch object exceeds max_size_bytes"
228		)
229	1	})?;
230	1.23k	hasher.update(&chunk);
231	1.23k	for (_, hasher8 ) in &mut checksum_hashers {
232	8	hasher.update(&chunk);
233	8	}
234	1.23k	file.write_all(&chunk).await ?0 ;
235		}
236	54	let digest = hasher.finalize_digest();
237	54	if let Some( alternatives24 ) = checksum {
238	24	let checksum_digests: Vec<_> = checksum_hashers
239	24	.into_iter()
240	24	.map(\|(function, mut hasher)\| ( function8 , hasher8 . finalize_digest8 ()))
241	24	.collect();
242	33	let matches24 = alternatives.iter()24 . any24 (\|expected\| {
243	33	let actual = if expected.function == digest_function {
244	23	Some(&digest)
245		} else {
246	10	checksum_digests
247	10	.iter()
248	10	.find(\|(function, _)\| *function == expected.function)
249	10	.map(\|(_, digest)\| digest)
250		};
251	33	actual.is_some_and(\|actual\| {
252	33	actual.packed_hash().as_slice() == expected.hash.as_slice()
253	33	})
254	33	});
255	24	if !matches {
256	8	return Err(make_err!(Code::DataLoss, "Fetch checksum mismatch"));
257	16	}
258	30	}
259	46	file.rewind().await ?0 ;
260	46	let (mut tx, rx) = make_buf_channel_pair();
261	46	let replay = async {
262	46	let mut buffer = vec![0; 64 * 1024];
263		loop {
264	4.18k	let count4.18k = file.read(&mut buffer).await ?0 ;
265	4.18k	if count == 0 {
266	45	break;
267	4.14k	}
268	4.14k	tx.send(Bytes::copy_from_slice(&buffer[..count])).await ?0 ;
269		}
270	45	tx.send_eof()
271	45	};
272	46	futures::try_join!(
273	46	replay,
274	46	self.cas.update(digest, rx, UploadSizeInfo::ExactSize(size))
275	0	)?;
276	45	Ok(digest)
277	57	}
278		}
279
280		#[derive(Debug, Clone)]
281		pub struct FetchServer {
282		stores: HashMap<String, FetchStoreInfo>,
283		}
284
285		impl FetchServer {
286	47	pub fn new(
287	47	configs: &[WithInstanceName<FetchConfig>],
288	47	store_manager: &StoreManager,
289	47	) -> Result<Self, Error> {
290	47	let mut stores = HashMap::with_capacity(configs.len());
291	48	for config in configs47 {
292	48	let store = store_manager
293	48	.get_store(&config.fetch_store)
294	48	.ok_or_else(\|\| {0
295	0	make_input_err!("'fetch_store': '{}' does not exist", config.fetch_store)
296	0	})?;
297	48	let downloader45 = config
298	48	.downloader
299	48	.as_ref()
300	48	.map(\|spec\| Downloader::new43 ( spec43 , store_manager43 ). map43 (Arc::new))
301	48	.transpose() ?3 ;
302	45	stores.insert(
303	45	config.instance_name.clone(),
304	45	FetchStoreInfo { store, downloader },
305		);
306		}
307	44	Ok(Self { stores })
308	47	}
309
310	0	pub fn into_service(self) -> Server<Self> {
311	0	Server::new(self)
312	0	}
313
314	99	async fn inner_fetch_blob(
315	99	&self,
316	99	request: FetchBlobRequest,
317	99	) -> Result<Response<FetchBlobResponse>, Error> {
318	99	let instance_name = &request.instance_name;
319	99	let store_info = self.stores.get(instance_name).ok_or_else(\|\| {0
320	0	make_input_err!("'instance_name' not configured for '{instance_name}'")
321	0	})?;
322
323	99	if request.uris.is_empty() {
324	0	return Err(Error::new(
325	0	Code::InvalidArgument,
326	0	"No uris in fetch request".to_owned(),
327	0	));
328	99	}
329	99	let download_options83 = if let Some( downloader94 ) = &store_info.downloader {
330	94	let checksum78 = Self::checksum(&request, downloader) ?16 ;
331	78	let timeout = match request.timeout {
332	2	Some(value) => Duration::try_from(value)
333	2	.map_err(\|_\| make_input_err!0 ("Invalid fetch timeout")) ?0
334	2	.min(downloader.timeout),
335	76	None => downloader.timeout,
336		};
337	78	Some((downloader, checksum, timeout))
338		} else {
339	5	None
340		};
341	84	for uri in &request.uris83 {
342	84	let asset_request = RemoteAssetQuery::new(uri.clone(), request.qualifiers.clone());
343	84	let asset_digest = asset_request.digest();
344	84	let asset_response_possible = store_info
345	84	.store
346	84	.get_part_unchunked(asset_digest, 0, None)
347	84	.await;
348
349	84	info!(
350	84	digest = format!("{}", asset_digest),
351		"Looked up fetch asset"
352		);
353
354	84	if let Ok( asset_response_raw34 ) = asset_response_possible {
355	34	let asset_response33 = RemoteAssetArtifact::decode(asset_response_raw)
356	34	.err_tip(\|\| "Failed to decode stored RemoteAssetArtifact") ?1 ;
357	33	if let Some( downloader30 ) = &store_info.downloader {
358		// Legacy index keys concatenate fields without boundaries. A matching
359		// key alone does not prove that the record belongs to this request.
360	30	if asset_response.uri != asset_request.uri
361	29	\|\| asset_response.qualifiers != asset_request.qualifiers
362		{
363	2	continue;
364	28	}
365	28	let now: prost_types::Timestamp = SystemTime::now().into();
366	28	let unexpired = asset_response.expire_at.as_ref().is_none_or(\|expiry\| {0
367	0	(expiry.seconds, expiry.nanos) > (now.seconds, now.nanos)
368	0	});
369	28	let stored_function =
370	28	DigestHasherFunc::try_from(asset_response.digest_function) ?0 ;
371	28	let same_function = stored_function == digest_hasher_func_from_context();
372	28	let fresh_enough =
373	28	request
374	28	.oldest_content_accepted
375	28	.as_ref()
376	28	.is_none_or(\|oldest\| {7
377	7	asset_response.fetched_at.as_ref().is_some_and(\|fetched\| {6
378	6	(fetched.seconds, fetched.nanos)
379	6	>= (oldest.seconds, oldest.nanos)
380	6	})
381	7	});
382	28	if !unexpired \|\| !same_function \|\| !fresh_enough {
383	4	continue;
384	24	}
385	24	let Some(blob) = asset_response.blob_digest.as_ref() else {
386	0	continue;
387		};
388	24	if downloader
389	24	.cas
390	24	.has(DigestInfo::try_from(blob) ?0 )
391	24	.await ?0
392	24	.is_none()
393		{
394	2	continue;
395	22	}
396	3	}
397	25	return Ok(Response::new(FetchBlobResponse {
398	25	status: Some(GoogleStatus {
399	25	code: Code::Ok.into(),
400	25	message: "Fetch object found".to_owned(),
401	25	details: vec![],
402	25	}),
403	25	uri: asset_response.uri,
404	25	qualifiers: asset_response.qualifiers,
405	25	expires_at: asset_response.expire_at,
406	25	blob_digest: asset_response.blob_digest,
407	25	digest_function: asset_response.digest_function,
408	25	}));
409	50	}
410		}
411	57	let Some(( downloader56 , checksum56 , timeout56 )) = download_options else {
412	1	let mut response =
413	1	Self::miss_response(&request, make_err!(Code::NotFound, "No item found"));
414	1	response.digest_function = default_digest_hasher_func().proto_digest_func().into();
415	1	return Ok(Response::new(response));
416		};
417	56	let retrieve = async {
418	56	let _slot = downloader
419	56	.slots
420	56	.acquire()
421	56	.await
422	56	.map_err(\|_\| make_err!0 ( Code::Unavailable0 , "Fetch downloader closed")) ?0 ;
423	56	let mut last_error = make_err!(Code::NotFound, "No downloadable URI found");
424	57	for uri in &request.uris56 {
425	57	let fetched_at = SystemTime::now().into();
426	57	match downloader.download(uri, checksum.as_deref()).await {
427	45	Ok(digest) => {
428	45	let mut artifact = RemoteAssetArtifact::new(
429	45	uri.clone(),
430	45	request.qualifiers.clone(),
431	45	digest.into(),
432	45	None,
433	45	digest_hasher_func_from_context().proto_digest_func().into(),
434		);
435	45	artifact.fetched_at = Some(fetched_at);
436	45	store_info
437	45	.store
438	45	.update_oneshot(artifact.digest(), artifact.as_bytes())
439	45	.await ?0 ;
440	45	return Ok(Response::new(FetchBlobResponse {
441	45	status: Some(GoogleStatus {
442	45	code: 0,
443	45	message: "Fetch object created".into(),
444	45	details: vec![],
445	45	}),
446	45	uri: artifact.uri,
447	45	qualifiers: artifact.qualifiers,
448	45	expires_at: artifact.expire_at,
449	45	blob_digest: artifact.blob_digest,
450	45	digest_function: artifact.digest_function,
451	45	}));
452		}
453	11	Err(error) => last_error = error,
454		}
455		}
456	10	Ok(Response::new(Self::miss_response(&request, last_error)))
457	55	};
458	56	tokio::time::timeout(timeout, retrieve)
459	56	.await
460	56	.unwrap_or_else(\|_\| {1
461	1	Ok(Response::new(Self::miss_response(
462	1	&request,
463	1	make_err!(Code::DeadlineExceeded, "Fetch origin timeout exceeded"),
464	1	)))
465	1	})
466	99	}
467
468	94	fn checksum(
469	94	request: &FetchBlobRequest,
470	94	downloader: &Downloader,
471	94	) -> Result<Option<Vec<ExpectedChecksum>>, Error> {
472	94	let mut checksum = None;
473	94	for qualifier60 in &request.qualifiers {
474	60	if qualifier.name.starts_with("http_header:")
475	57	\|\| qualifier.name.starts_with("http_header_url:")
476		{
477	5	return Err(make_input_err!(
478	5	"HTTP header qualifiers are not supported by the fetch downloader"
479	5	));
480	55	}
481	55	if qualifier.name != "checksum.sri" {
482	11	if qualifier.name != "bazel.canonical_id" && !downloader.allow_unknown_qualifiers {
483	2	return Err(make_input_err!("Unsupported fetch qualifier"));
484	9	}
485	9	continue;
486	44	}
487	44	if checksum.is_some() {
488	1	return Err(make_input_err!("Duplicate checksum qualifier"));
489	43	}
490	43	let mut alternatives = Vec::new();
491	59	for token in qualifier.value.split_whitespace()43 {
492	59	let ( algorithm58 , encoded58 ) = token
493	59	.split_once('-')
494	59	.ok_or_else(\|\| make_input_err!1 ("Invalid checksum.sri format")) ?1 ;
495	58	if algorithm.is_empty() \|\| encoded.is_empty() {
496	0	return Err(make_input_err!("Invalid checksum.sri format"));
497	58	}
498	58	let hash56 = BASE64
499	58	.decode(encoded)
500	58	.map_err(\|_\| make_input_err!2 ("Invalid checksum.sri encoding")) ?2 ;
501		// Unsupported algorithms may coexist with supported alternatives.
502	56	let Ok( function54 ) = DigestHasherFunc::try_from(algorithm) else {
503	2	continue;
504		};
505	54	let hash53 = hash
506	54	.try_into()
507	54	.map_err(\|_\| make_input_err!1 ("Invalid checksum.sri length")) ?1 ;
508	53	alternatives.push(ExpectedChecksum { function, hash });
509		}
510	39	if alternatives.is_empty() {
511	3	return Err(make_input_err!("No supported checksum.sri alternative"));
512	36	}
513	36	checksum = Some(alternatives);
514		}
515	79	if downloader.require_checksum && checksum2 . is_none2 () {
516	1	return Err(make_input_err!("Fetch downloader requires checksum.sri"));
517	78	}
518	78	Ok(checksum)
519	94	}
520
521	12	fn miss_response(request: &FetchBlobRequest, error: Error) -> FetchBlobResponse {
522	12	FetchBlobResponse {
523	12	status: Some(error.into()),
524	12	uri: request.uris.first().cloned().unwrap_or_default(),
525	12	qualifiers: vec![],
526	12	expires_at: None,
527	12	blob_digest: None,
528	12	digest_function: digest_hasher_func_from_context().proto_digest_func().into(),
529	12	}
530	12	}
531		}
532
533		#[tonic::async_trait]
534		impl Fetch for FetchServer {
535		#[allow(clippy::blocks_in_conditions)]
536		#[instrument(
537		err(level = Level::WARN),
538		skip_all,
539		fields(
540		uri_count = grpc_request.get_ref().uris.len(),
541		qualifier_count = grpc_request.get_ref().qualifiers.len(),
542		digest_function = grpc_request.get_ref().digest_function,
543		)
544		)]
545		async fn fetch_blob(
546		&self,
547		grpc_request: Request<FetchBlobRequest>,
548		) -> Result<Response<FetchBlobResponse>, Status> {
549		let request = grpc_request.into_inner();
550		let digest_function = request.digest_function;
551		self.inner_fetch_blob(request)
552		.instrument(error_span!("fetch_server_fetch_blob"))
553		.with_context(
554		make_ctx_for_hash_func(digest_function).err_tip(\|\| "In FetchServer::fetch_blob")?,
555		)
556		.await
557		.err_tip(\|\| "Failed on fetch_blob() command")
558		.map_err(Into::into)
559		}
560
561		#[allow(clippy::blocks_in_conditions)]
562		#[instrument(
563		err(level = Level::WARN),
564		skip_all,
565		fields(
566		uri_count = _grpc_request.get_ref().uris.len(),
567		qualifier_count = _grpc_request.get_ref().qualifiers.len(),
568		digest_function = _grpc_request.get_ref().digest_function,
569		)
570		)]
571		async fn fetch_directory(
572		&self,
573		_grpc_request: Request<FetchDirectoryRequest>,
574		) -> Result<Response<FetchDirectoryResponse>, Status> {
575		Err(Status::unimplemented("FetchDirectory not implemented"))
576		}
577		}