19676380878

Committed 25 Nov 2025 04:17PM UTC coverage: 54.592% (-0.1%) from 54.722%

Build # 19676380878

Build Type

push

github

Committed by

0xmichalis

Commit Message

feat(server): dynamic quotes based on total backup size

Run Details

89 of 153 new or added lines in 6 files covered. (58.17%)

130 existing lines in 5 files now uncovered.

3222 of 5902 relevant lines covered (54.59%)

10.57 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.98

/src/httpclient/fetch.rs

use reqwest::{header::CONTENT_LENGTH, Method};
use tracing::{info, warn};

use crate::ipfs::config::IpfsGatewayConfig;
use crate::ipfs::url::{generate_url_for_gateways, GatewayUrl};

async fn send_request(
    method: Method,
    url: &str,
    bearer_token: Option<String>,
) -> anyhow::Result<reqwest::Response> {
    let client = reqwest::Client::builder()
        .user_agent(crate::USER_AGENT)
        .build()?;

    let mut req = client.request(method, url);
    if let Some(token) = bearer_token {
        req = req.bearer_auth(token);
    }

    Ok(req.send().await?)
}

pub(crate) async fn fetch_url(
    url: &str,
    bearer_token: Option<String>,
) -> anyhow::Result<reqwest::Response> {
    send_request(Method::GET, url, bearer_token).await
}

pub(crate) async fn head_url(
    url: &str,
    bearer_token: Option<String>,
) -> anyhow::Result<reqwest::Response> {
    send_request(Method::HEAD, url, bearer_token).await
}

pub(crate) fn create_http_error(status: reqwest::StatusCode, url: &str) -> anyhow::Error {
    anyhow::anyhow!("HTTP error: status {status} from {url}")
}

pub(crate) async fn try_fetch_response(
    url: &str,
    gateways: &[IpfsGatewayConfig],
) -> (
    anyhow::Result<reqwest::Response>,
    Option<reqwest::StatusCode>,
) {
    match fetch_url(url, None).await {
        Ok(response) => {
            let status = response.status();
            if status.is_success() {
                (Ok(response), Some(status))
            } else {
                let error = create_http_error(status, url);
                match retry_with_gateways(url, error, gateways).await {
                    Ok(alt_response) => {
                        let alt_status = alt_response.status();
                        if alt_status.is_success() {
                            (Ok(alt_response), Some(alt_status))
                        } else {
                            (
                                Err(create_http_error(alt_status, alt_response.url().as_str())),
                                Some(alt_status),
                            )
                        }
                    }
                    Err(gateway_err) => (Err(gateway_err), Some(status)),
                }
            }
        }
        Err(e) => match retry_with_gateways(url, e, gateways).await {
            Ok(response) => {
                let status = response.status();
                if status.is_success() {
                    (Ok(response), Some(status))
                } else {
                    (Err(create_http_error(status, url)), Some(status))
                }
            }
            Err(gateway_err) => (Err(gateway_err), None),
        },
    }
}

/// Retry a failed HTTP request against configured IPFS gateways using the
/// provided HTTP method so both GET and HEAD flows can share the same logic.
/// The original error is logged for context and returned if no gateway
/// succeeds.
async fn retry_with_gateways_with_method(
    url: &str,
    original_error: anyhow::Error,
    gateways: &[IpfsGatewayConfig],
    method: Method,
) -> anyhow::Result<reqwest::Response> {
    let gateway_urls = match generate_url_for_gateways(url, gateways) {
        Some(urls) => {
            warn!(
                "IPFS gateway error for {}, retrying with other gateways: {}",
                url, original_error
            );
            urls
        }
        None => return Err(original_error),
    };

    let mut last_err = original_error;
    let alternative_gateways: Vec<_> = gateway_urls
        .into_iter()
        .filter(|gateway_url| gateway_url.url != url)
        .collect();

    for GatewayUrl {
        url: new_url,
        bearer_token,
    } in alternative_gateways
    {
        match send_request(method.clone(), &new_url, bearer_token).await {
            Ok(response) => {
                let status = response.status();
                if status.is_success() {
                    info!(
                        "Received successful response from alternative IPFS gateway: {} (status: {})",
                        new_url, status
                    );
                    return Ok(response);
                } else {
                    warn!("Alternative IPFS gateway {new_url} failed: {status}");
                    last_err = anyhow::anyhow!("{status}");
                }
            }
            Err(err) => {
                warn!("Failed to fetch from IPFS gateway {}: {}", new_url, err);
                last_err = err;
            }
        }
    }

    Err(last_err)
}

pub(crate) async fn retry_with_gateways(
    url: &str,
    original_error: anyhow::Error,
    gateways: &[IpfsGatewayConfig],
) -> anyhow::Result<reqwest::Response> {
    retry_with_gateways_with_method(url, original_error, gateways, Method::GET).await
}

fn extract_content_length(response: &reqwest::Response) -> Option<u64> {
    response
        .headers()
        .get(CONTENT_LENGTH)
        .and_then(|h| h.to_str().ok())
        .and_then(|s| s.parse::<u64>().ok())
        .or_else(|| response.content_length())
}

/// Attempt to read the `Content-Length` of `url` using HEAD requests, falling
/// back to alternate IPFS gateways when available. Returns the parsed byte size
/// and the HTTP status code (if any) encountered while probing.
pub(crate) async fn try_head_content_length(
    url: &str,
    gateways: &[IpfsGatewayConfig],
) -> (anyhow::Result<u64>, Option<reqwest::StatusCode>) {
    match head_url(url, None).await {
        Ok(response) => {
            let status = response.status();
            if status.is_success() {
                if let Some(len) = extract_content_length(&response) {
                    (Ok(len), Some(status))
                } else {
                    (
                        Err(anyhow::anyhow!(
                            "HEAD response missing Content-Length for {}",
                            url
                        )),
                        Some(status),
                    )
                }
            } else {
                let error = create_http_error(status, url);
                match retry_with_gateways_with_method(url, error, gateways, Method::HEAD).await {
                    Ok(alt_response) => {
                        let alt_status = alt_response.status();
                        if alt_status.is_success() {
                            if let Some(len) = extract_content_length(&alt_response) {
                                (Ok(len), Some(alt_status))
                            } else {
                                (
                                    Err(anyhow::anyhow!(
                                        "HEAD response missing Content-Length for {}",
                                        alt_response.url()
                                    )),
                                    Some(alt_status),
                                )
                            }
                        } else {
                            (
                                Err(create_http_error(alt_status, alt_response.url().as_str())),
                                Some(alt_status),
                            )
                        }
                    }
                    Err(gateway_err) => (Err(gateway_err), Some(status)),
                }
            }
        }
        Err(e) => match retry_with_gateways_with_method(url, e, gateways, Method::HEAD).await {
            Ok(response) => {
                let status = response.status();
                if status.is_success() {
                    if let Some(len) = extract_content_length(&response) {
                        (Ok(len), Some(status))
                    } else {
                        (
                            Err(anyhow::anyhow!(
                                "HEAD response missing Content-Length for {}",
                                response.url()
                            )),
                            Some(status),
                        )
                    }
                } else {
                    (Err(create_http_error(status, url)), Some(status))
                }
            }
            Err(gateway_err) => (Err(gateway_err), None),
        },
    }
}

#[cfg(test)]
mod try_fetch_response_tests {
    use wiremock::{
        matchers::{method, path},
        Mock, MockServer, ResponseTemplate,
    };

    use crate::ipfs::config::{IpfsGatewayConfig, IpfsGatewayType};

    fn leak_str(s: String) -> &'static str {
        Box::leak(s.into_boxed_str())
    }

    fn gw(base: &str) -> IpfsGatewayConfig {
        IpfsGatewayConfig {
            url: leak_str(base.to_string()),
            gateway_type: IpfsGatewayType::Path,
            bearer_token_env: None,
        }
    }

    #[tokio::test]
    async fn test_http_200_success() {
        let mock_server = MockServer::start().await;
        let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
        let url = format!("{}/ipfs/{}", mock_server.uri(), cid);

        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", cid)))
            .respond_with(ResponseTemplate::new(200).set_body_string("Hello, World!"))
            .mount(&mock_server)
            .await;

        let (res, status) = super::try_fetch_response(&url, &[]).await;
        assert!(status.is_some());
        assert_eq!(status.unwrap(), reqwest::StatusCode::OK);
        assert!(res.is_ok());
        let body = res.unwrap().bytes().await.unwrap();
        assert_eq!(body.as_ref(), b"Hello, World!");
    }

    #[tokio::test]
    async fn test_http_404_no_retry() {
        let mock_server = MockServer::start().await;
        let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
        let url = format!("{}/ipfs/{}", mock_server.uri(), cid);

        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", cid)))
            .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
            .expect(1)
            .mount(&mock_server)
            .await;

        let (res, status) = super::try_fetch_response(&url, &[]).await;
        assert!(status.is_some());
        assert_eq!(status.unwrap(), reqwest::StatusCode::NOT_FOUND);
        assert!(res.is_err());
        let err = res.err().unwrap().to_string();
        assert!(err.contains("HTTP error: status 404"));
    }

    #[tokio::test]
    async fn test_http_500_retry_and_fail() {
        let mock_server = MockServer::start().await;
        let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
        let url = format!("{}/ipfs/{}", mock_server.uri(), cid);

        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", cid)))
            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
            .expect(1)
            .mount(&mock_server)
            .await;

        let (res, status) = super::try_fetch_response(&url, &[]).await;
        assert!(status.is_some());
        assert_eq!(status.unwrap(), reqwest::StatusCode::INTERNAL_SERVER_ERROR);
        assert!(res.is_err());
        let err = res.err().unwrap().to_string();
        assert!(err.contains("HTTP error: status 500"));
    }

    #[tokio::test]
    async fn test_initial_error_then_alternative_gateway_succeeds() {
        // Arrange: original gateway fails, alternative returns 200
        let server_a = MockServer::start().await;
        let server_b = MockServer::start().await;
        let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
        let content_path = format!("{}/ok", cid);
        let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);

        // A returns 500 (server error)
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(500).set_body_string("Server Error"))
            .mount(&server_a)
            .await;

        // B returns 200 for the same IPFS path
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(200).set_body_string("ALT OK"))
            .mount(&server_b)
            .await;

        // Gateways: first the failing one (A), then the working mock (B)
        let gateways = vec![gw(&server_a.uri()), gw(&server_b.uri())];

        // Act
        let (res, status) = super::try_fetch_response(&original_url, &gateways).await;

        // Assert
        assert!(res.is_ok());
        assert_eq!(status, Some(reqwest::StatusCode::OK));
        let body = res.unwrap().bytes().await.unwrap();
        assert_eq!(body.as_ref(), b"ALT OK");
    }

    #[tokio::test]
    async fn test_initial_network_error_then_alternative_gateway_succeeds() {
        // Arrange: original gateway has network error, alternative returns 200
        let server_b = MockServer::start().await;
        let dead_gateway_base = "http://127.0.0.1:9"; // simulate connect error
        let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
        let content_path = format!("{}/network_test", cid);
        let original_url = format!("{}/ipfs/{}", dead_gateway_base, content_path);

        // B returns 200 for the same IPFS path
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(200).set_body_string("NETWORK RETRY OK"))
            .mount(&server_b)
            .await;

        // Gateways: first the dead one (original), then the working mock
        let gateways = vec![gw(dead_gateway_base), gw(&server_b.uri())];

        // Act
        let (res, status) = super::try_fetch_response(&original_url, &gateways).await;

        // Assert
        assert!(res.is_ok());
        assert_eq!(status, Some(reqwest::StatusCode::OK));
        let body = res.unwrap().bytes().await.unwrap();
        assert_eq!(body.as_ref(), b"NETWORK RETRY OK");
    }
}

#[cfg(test)]
mod retry_with_gateways_tests {
    use wiremock::{
        matchers::{method, path},
        Mock, MockServer, ResponseTemplate,
    };

    use crate::ipfs::config::{IpfsGatewayConfig, IpfsGatewayType};

    // Helper to leak a String into a 'static str for IpfsGatewayConfig
    fn leak_str(s: String) -> &'static str {
        Box::leak(s.into_boxed_str())
    }

    // Build a path-based gateway config from a base URL
    fn gw(base: &str) -> IpfsGatewayConfig {
        IpfsGatewayConfig {
            url: leak_str(base.to_string()),
            gateway_type: IpfsGatewayType::Path,
            bearer_token_env: None,
        }
    }

    #[tokio::test]
    async fn first_alternative_succeeds() {
        // Arrange gateways A(original), B(success), C(not used)
        let server_a = MockServer::start().await;
        let server_b = MockServer::start().await;
        let server_c = MockServer::start().await;

        let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/ok";
        let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);

        // B returns 200
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(200).set_body_string("B OK"))
            .mount(&server_b)
            .await;

        let gateways = vec![
            gw(&server_a.uri()),
            gw(&server_b.uri()),
            gw(&server_c.uri()),
        ];

        // Act
        let res =
            super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;

        // Assert
        assert!(res.is_ok());
        let body = res.unwrap().bytes().await.unwrap();
        assert_eq!(body.as_ref(), b"B OK");
    }

    #[tokio::test]
    async fn first_alternative_http_error_second_succeeds() {
        // Arrange gateways A(original), B(500), C(200)
        let server_a = MockServer::start().await;
        let server_b = MockServer::start().await;
        let server_c = MockServer::start().await;

        let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/ok2";
        let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);

        // B returns 500
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(500))
            .mount(&server_b)
            .await;

        // C returns 200
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(200).set_body_string("C OK"))
            .mount(&server_c)
            .await;

        let gateways = vec![
            gw(&server_a.uri()),
            gw(&server_b.uri()),
            gw(&server_c.uri()),
        ];

        // Act
        let res =
            super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;

        // Assert
        assert!(res.is_ok());
        let body = res.unwrap().bytes().await.unwrap();
        assert_eq!(body.as_ref(), b"C OK");
    }

    #[tokio::test]
    async fn first_alternative_non_http_error_second_succeeds() {
        // Arrange gateways A(original), B(non-http error), C(200)
        let server_a = MockServer::start().await;
        let server_c = MockServer::start().await;

        // Use an unroutable/closed port for B to simulate network error
        let dead_gateway_base = "http://127.0.0.1:9"; // port 9 (discard) likely closed

        let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/ok3";
        let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);

        // C returns 200
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(200).set_body_string("C OK"))
            .mount(&server_c)
            .await;

        let gateways = vec![
            gw(&server_a.uri()),
            gw(dead_gateway_base),
            gw(&server_c.uri()),
        ];

        // Act
        let res =
            super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;

        // Assert
        assert!(res.is_ok());
        let body = res.unwrap().bytes().await.unwrap();
        assert_eq!(body.as_ref(), b"C OK");
    }

    #[tokio::test]
    async fn both_alternatives_fail() {
        // Arrange gateways A(original), B(500), C(502)
        let server_a = MockServer::start().await;
        let server_b = MockServer::start().await;
        let server_c = MockServer::start().await;

        let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/fail";
        let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);

        // B returns 500
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(500))
            .mount(&server_b)
            .await;

        // C returns 502
        Mock::given(method("GET"))
            .and(path(format!("/ipfs/{}", content_path)))
            .respond_with(ResponseTemplate::new(502))
            .mount(&server_c)
            .await;

        let gateways = vec![
            gw(&server_a.uri()),
            gw(&server_b.uri()),
            gw(&server_c.uri()),
        ];

        // Act
        let res =
            super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;

        // Assert: error should reflect the last error (502)
        assert!(res.is_err());
        let err = res.err().unwrap().to_string();
        assert!(err.contains("502"));
    }

    #[tokio::test]
    async fn non_ipfs_url_returns_original_error_without_warning() {
        // Test that non-IPFS URLs return the original error without logging IPFS gateway warnings
        let non_ipfs_url =
            "https://mirror.xyz/10/0xceda033195af537e16d203a4d7fbfe1c5f0eb843/render";
        let original_error_msg = "HTTP error: status 429 Too Many Requests";
        let original_error = anyhow::anyhow!(original_error_msg);
        let gateways = vec![gw("https://ipfs.io")];

        // Act
        let res = super::retry_with_gateways(non_ipfs_url, original_error, &gateways).await;

        // Assert: should return the original error without trying IPFS gateways
        assert!(res.is_err());
        let err = res.err().unwrap();
        assert_eq!(err.to_string(), original_error_msg);
    }
}

#[cfg(test)]
mod fetch_url_tests {
    use wiremock::{
        matchers::{header, method, path},
        Mock, MockServer, ResponseTemplate,
    };

    #[tokio::test]
    async fn test_fetch_url_without_bearer_token() {
        let mock_server = MockServer::start().await;
        let url = format!("{}/test", mock_server.uri());

        Mock::given(method("GET"))
            .and(path("/test"))
            .and(header("User-Agent", crate::USER_AGENT))
            .respond_with(ResponseTemplate::new(200).set_body_string("OK"))
            .mount(&mock_server)
            .await;

        let result = super::fetch_url(&url, None).await;
        assert!(result.is_ok());
        let response = result.unwrap();
        assert_eq!(response.status(), reqwest::StatusCode::OK);
    }

    #[tokio::test]
    async fn test_fetch_url_with_bearer_token() {
        let mock_server = MockServer::start().await;
        let url = format!("{}/protected", mock_server.uri());
        let token = "test-bearer-token-123";
        let auth_header = format!("Bearer {}", token);

        Mock::given(method("GET"))
            .and(path("/protected"))
            .and(header("User-Agent", crate::USER_AGENT))
            .and(header("Authorization", auth_header.as_str()))
            .respond_with(ResponseTemplate::new(200).set_body_string("Protected OK"))
            .mount(&mock_server)
            .await;

        let result = super::fetch_url(&url, Some(token.to_string())).await;
        assert!(result.is_ok());
        let response = result.unwrap();
        assert_eq!(response.status(), reqwest::StatusCode::OK);
    }

    #[tokio::test]
    async fn test_fetch_url_with_empty_bearer_token() {
        let mock_server = MockServer::start().await;
        let url = format!("{}/test", mock_server.uri());

        Mock::given(method("GET"))
            .and(path("/test"))
            .and(header("User-Agent", crate::USER_AGENT))
            .respond_with(ResponseTemplate::new(200).set_body_string("OK"))
            .mount(&mock_server)
            .await;

        // Empty string should not add Authorization header
        let result = super::fetch_url(&url, Some("".to_string())).await;
        assert!(result.is_ok());
        let response = result.unwrap();
        assert_eq!(response.status(), reqwest::StatusCode::OK);
    }

    #[tokio::test]
    async fn test_fetch_url_network_error() {
        // Use a non-existent URL to test network error handling
        let url = "http://127.0.0.1:9/nonexistent";

        let result = super::fetch_url(url, None).await;
        assert!(result.is_err());
    }
}

1	use reqwest::{header::CONTENT_LENGTH, Method};
2	use tracing::{info, warn};
3
4	use crate::ipfs::config::IpfsGatewayConfig;
5	use crate::ipfs::url::{generate_url_for_gateways, GatewayUrl};
6
7	async fn send_request(	37✔
8	method: Method,
9	url: &str,
10	bearer_token: Option<String>,
11	) -> anyhow::Result<reqwest::Response> {
12	let client = reqwest::Client::builder()	74✔
13	.user_agent(crate::USER_AGENT)	37✔
14	.build()?;
15
16	let mut req = client.request(method, url);	185✔
17	if let Some(token) = bearer_token {	41✔
18	req = req.bearer_auth(token);	6✔
19	}
20
21	Ok(req.send().await?)	78✔
22	}
23
24	pub(crate) async fn fetch_url(	27✔
25	url: &str,
26	bearer_token: Option<String>,
27	) -> anyhow::Result<reqwest::Response> {
28	send_request(Method::GET, url, bearer_token).await	108✔
29	}
30
31	pub(crate) async fn head_url(	1✔
32	url: &str,
33	bearer_token: Option<String>,
34	) -> anyhow::Result<reqwest::Response> {
35	send_request(Method::HEAD, url, bearer_token).await	4✔
36	}
37
38	pub(crate) fn create_http_error(status: reqwest::StatusCode, url: &str) -> anyhow::Error {	13✔
39	anyhow::anyhow!("HTTP error: status {status} from {url}")	26✔
40	}
41
42	pub(crate) async fn try_fetch_response(	23✔
43	url: &str,
44	gateways: &[IpfsGatewayConfig],
45	) -> (
46	anyhow::Result<reqwest::Response>,
47	Option<reqwest::StatusCode>,
48	) {
49	match fetch_url(url, None).await {	92✔
50	Ok(response) => {	21✔
51	let status = response.status();	63✔
52	if status.is_success() {	42✔
53	(Ok(response), Some(status))	8✔
54	} else {
55	let error = create_http_error(status, url);	52✔
56	match retry_with_gateways(url, error, gateways).await {	65✔
57	Ok(alt_response) => {	1✔
58	let alt_status = alt_response.status();	3✔
59	if alt_status.is_success() {	2✔
60	(Ok(alt_response), Some(alt_status))	1✔
61	} else {
62	(
UNCOV 63	Err(create_http_error(alt_status, alt_response.url().as_str())),	×
UNCOV 64	Some(alt_status),	×
65	)
66	}
67	}
68	Err(gateway_err) => (Err(gateway_err), Some(status)),	24✔
69	}
70	}
71	}
72	Err(e) => match retry_with_gateways(url, e, gateways).await {	12✔
73	Ok(response) => {	1✔
74	let status = response.status();	3✔
75	if status.is_success() {	2✔
76	(Ok(response), Some(status))	1✔
77	} else {
78	(Err(create_http_error(status, url)), Some(status))	×
79	}
80	}
81	Err(gateway_err) => (Err(gateway_err), None),	2✔
82	},
83	}
84	}
85
86	/// Retry a failed HTTP request against configured IPFS gateways using the
87	/// provided HTTP method so both GET and HEAD flows can share the same logic.
88	/// The original error is logged for context and returned if no gateway
89	/// succeeds.
90	async fn retry_with_gateways_with_method(	20✔
91	url: &str,
92	original_error: anyhow::Error,
93	gateways: &[IpfsGatewayConfig],
94	method: Method,
95	) -> anyhow::Result<reqwest::Response> {
96	let gateway_urls = match generate_url_for_gateways(url, gateways) {	48✔
97	Some(urls) => {	8✔
98	warn!(	8✔
UNCOV 99	"IPFS gateway error for {}, retrying with other gateways: {}",	×
100	url, original_error
101	);
102	urls	8✔
103	}
104	None => return Err(original_error),	12✔
105	};
106
107	let mut last_err = original_error;	16✔
108	let alternative_gateways: Vec<_> = gateway_urls	24✔
109	.into_iter()
110	.filter(\|gateway_url\| gateway_url.url != url)	40✔
111	.collect();
112
113	for GatewayUrl {
114	url: new_url,	9✔
115	bearer_token,	9✔
116	} in alternative_gateways	12✔
117	{
118	match send_request(method.clone(), &new_url, bearer_token).await {	54✔
119	Ok(response) => {	8✔
120	let status = response.status();	24✔
121	if status.is_success() {	16✔
122	info!(	5✔
UNCOV 123	"Received successful response from alternative IPFS gateway: {} (status: {})",	×
124	new_url, status
125	);
126	return Ok(response);	5✔
127	} else {
128	warn!("Alternative IPFS gateway {new_url} failed: {status}");	3✔
129	last_err = anyhow::anyhow!("{status}");	9✔
130	}
131	}
132	Err(err) => {	1✔
133	warn!("Failed to fetch from IPFS gateway {}: {}", new_url, err);	1✔
134	last_err = err;	2✔
135	}
136	}
137	}
138
139	Err(last_err)	3✔
140	}
141
142	pub(crate) async fn retry_with_gateways(	20✔
143	url: &str,
144	original_error: anyhow::Error,
145	gateways: &[IpfsGatewayConfig],
146	) -> anyhow::Result<reqwest::Response> {
147	retry_with_gateways_with_method(url, original_error, gateways, Method::GET).await	100✔
148	}
149
150	fn extract_content_length(response: &reqwest::Response) -> Option<u64> {	1✔
151	response	1✔
152	.headers()
153	.get(CONTENT_LENGTH)	2✔
154	.and_then(\|h\| h.to_str().ok())	4✔
155	.and_then(\|s\| s.parse::<u64>().ok())	4✔
156	.or_else(\|\| response.content_length())	1✔
157	}
158
159	/// Attempt to read the `Content-Length` of `url` using HEAD requests, falling
160	/// back to alternate IPFS gateways when available. Returns the parsed byte size
161	/// and the HTTP status code (if any) encountered while probing.
162	pub(crate) async fn try_head_content_length(	1✔
163	url: &str,
164	gateways: &[IpfsGatewayConfig],
165	) -> (anyhow::Result<u64>, Option<reqwest::StatusCode>) {
166	match head_url(url, None).await {	4✔
167	Ok(response) => {	1✔
168	let status = response.status();	3✔
169	if status.is_success() {	2✔
170	if let Some(len) = extract_content_length(&response) {	2✔
171	(Ok(len), Some(status))	1✔
172	} else {
173	(
UNCOV 174	Err(anyhow::anyhow!(	×
UNCOV 175	"HEAD response missing Content-Length for {}",	×
UNCOV 176	url	×
177	)),
UNCOV 178	Some(status),	×
179	)
180	}
181	} else {
UNCOV 182	let error = create_http_error(status, url);	×
UNCOV 183	match retry_with_gateways_with_method(url, error, gateways, Method::HEAD).await {	×
UNCOV 184	Ok(alt_response) => {	×
UNCOV 185	let alt_status = alt_response.status();	×
UNCOV 186	if alt_status.is_success() {	×
UNCOV 187	if let Some(len) = extract_content_length(&alt_response) {	×
UNCOV 188	(Ok(len), Some(alt_status))	×
189	} else {
190	(
UNCOV 191	Err(anyhow::anyhow!(	×
UNCOV 192	"HEAD response missing Content-Length for {}",	×
UNCOV 193	alt_response.url()	×
194	)),
UNCOV 195	Some(alt_status),	×
196	)
197	}
198	} else {
199	(
UNCOV 200	Err(create_http_error(alt_status, alt_response.url().as_str())),	×
UNCOV 201	Some(alt_status),	×
202	)
203	}
204	}
UNCOV 205	Err(gateway_err) => (Err(gateway_err), Some(status)),	×
206	}
207	}
208	}
UNCOV 209	Err(e) => match retry_with_gateways_with_method(url, e, gateways, Method::HEAD).await {	×
UNCOV 210	Ok(response) => {	×
UNCOV 211	let status = response.status();	×
UNCOV 212	if status.is_success() {	×
UNCOV 213	if let Some(len) = extract_content_length(&response) {	×
UNCOV 214	(Ok(len), Some(status))	×
215	} else {
216	(
UNCOV 217	Err(anyhow::anyhow!(	×
UNCOV 218	"HEAD response missing Content-Length for {}",	×
UNCOV 219	response.url()	×
220	)),
UNCOV 221	Some(status),	×
222	)
223	}
224	} else {
UNCOV 225	(Err(create_http_error(status, url)), Some(status))	×
226	}
227	}
UNCOV 228	Err(gateway_err) => (Err(gateway_err), None),	×
229	},
230	}
231	}
232
233	#[cfg(test)]
234	mod try_fetch_response_tests {
235	use wiremock::{
236	matchers::{method, path},
237	Mock, MockServer, ResponseTemplate,
238	};
239
240	use crate::ipfs::config::{IpfsGatewayConfig, IpfsGatewayType};
241
242	fn leak_str(s: String) -> &'static str {
243	Box::leak(s.into_boxed_str())
244	}
245
246	fn gw(base: &str) -> IpfsGatewayConfig {
247	IpfsGatewayConfig {
248	url: leak_str(base.to_string()),
249	gateway_type: IpfsGatewayType::Path,
250	bearer_token_env: None,
251	}
252	}
253
254	#[tokio::test]
255	async fn test_http_200_success() {
256	let mock_server = MockServer::start().await;
257	let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
258	let url = format!("{}/ipfs/{}", mock_server.uri(), cid);
259
260	Mock::given(method("GET"))
261	.and(path(format!("/ipfs/{}", cid)))
262	.respond_with(ResponseTemplate::new(200).set_body_string("Hello, World!"))
263	.mount(&mock_server)
264	.await;
265
266	let (res, status) = super::try_fetch_response(&url, &[]).await;
267	assert!(status.is_some());
268	assert_eq!(status.unwrap(), reqwest::StatusCode::OK);
269	assert!(res.is_ok());
270	let body = res.unwrap().bytes().await.unwrap();
271	assert_eq!(body.as_ref(), b"Hello, World!");
272	}
273
274	#[tokio::test]
275	async fn test_http_404_no_retry() {
276	let mock_server = MockServer::start().await;
277	let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
278	let url = format!("{}/ipfs/{}", mock_server.uri(), cid);
279
280	Mock::given(method("GET"))
281	.and(path(format!("/ipfs/{}", cid)))
282	.respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
283	.expect(1)
284	.mount(&mock_server)
285	.await;
286
287	let (res, status) = super::try_fetch_response(&url, &[]).await;
288	assert!(status.is_some());
289	assert_eq!(status.unwrap(), reqwest::StatusCode::NOT_FOUND);
290	assert!(res.is_err());
291	let err = res.err().unwrap().to_string();
292	assert!(err.contains("HTTP error: status 404"));
293	}
294
295	#[tokio::test]
296	async fn test_http_500_retry_and_fail() {
297	let mock_server = MockServer::start().await;
298	let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
299	let url = format!("{}/ipfs/{}", mock_server.uri(), cid);
300
301	Mock::given(method("GET"))
302	.and(path(format!("/ipfs/{}", cid)))
303	.respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
304	.expect(1)
305	.mount(&mock_server)
306	.await;
307
308	let (res, status) = super::try_fetch_response(&url, &[]).await;
309	assert!(status.is_some());
310	assert_eq!(status.unwrap(), reqwest::StatusCode::INTERNAL_SERVER_ERROR);
311	assert!(res.is_err());
312	let err = res.err().unwrap().to_string();
313	assert!(err.contains("HTTP error: status 500"));
314	}
315
316	#[tokio::test]
317	async fn test_initial_error_then_alternative_gateway_succeeds() {
318	// Arrange: original gateway fails, alternative returns 200
319	let server_a = MockServer::start().await;
320	let server_b = MockServer::start().await;
321	let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
322	let content_path = format!("{}/ok", cid);
323	let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);
324
325	// A returns 500 (server error)
326	Mock::given(method("GET"))
327	.and(path(format!("/ipfs/{}", content_path)))
328	.respond_with(ResponseTemplate::new(500).set_body_string("Server Error"))
329	.mount(&server_a)
330	.await;
331
332	// B returns 200 for the same IPFS path
333	Mock::given(method("GET"))
334	.and(path(format!("/ipfs/{}", content_path)))
335	.respond_with(ResponseTemplate::new(200).set_body_string("ALT OK"))
336	.mount(&server_b)
337	.await;
338
339	// Gateways: first the failing one (A), then the working mock (B)
340	let gateways = vec![gw(&server_a.uri()), gw(&server_b.uri())];
341
342	// Act
343	let (res, status) = super::try_fetch_response(&original_url, &gateways).await;
344
345	// Assert
346	assert!(res.is_ok());
347	assert_eq!(status, Some(reqwest::StatusCode::OK));
348	let body = res.unwrap().bytes().await.unwrap();
349	assert_eq!(body.as_ref(), b"ALT OK");
350	}
351
352	#[tokio::test]
353	async fn test_initial_network_error_then_alternative_gateway_succeeds() {
354	// Arrange: original gateway has network error, alternative returns 200
355	let server_b = MockServer::start().await;
356	let dead_gateway_base = "http://127.0.0.1:9"; // simulate connect error
357	let cid = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco";
358	let content_path = format!("{}/network_test", cid);
359	let original_url = format!("{}/ipfs/{}", dead_gateway_base, content_path);
360
361	// B returns 200 for the same IPFS path
362	Mock::given(method("GET"))
363	.and(path(format!("/ipfs/{}", content_path)))
364	.respond_with(ResponseTemplate::new(200).set_body_string("NETWORK RETRY OK"))
365	.mount(&server_b)
366	.await;
367
368	// Gateways: first the dead one (original), then the working mock
369	let gateways = vec![gw(dead_gateway_base), gw(&server_b.uri())];
370
371	// Act
372	let (res, status) = super::try_fetch_response(&original_url, &gateways).await;
373
374	// Assert
375	assert!(res.is_ok());
376	assert_eq!(status, Some(reqwest::StatusCode::OK));
377	let body = res.unwrap().bytes().await.unwrap();
378	assert_eq!(body.as_ref(), b"NETWORK RETRY OK");
379	}
380	}
381
382	#[cfg(test)]
383	mod retry_with_gateways_tests {
384	use wiremock::{
385	matchers::{method, path},
386	Mock, MockServer, ResponseTemplate,
387	};
388
389	use crate::ipfs::config::{IpfsGatewayConfig, IpfsGatewayType};
390
391	// Helper to leak a String into a 'static str for IpfsGatewayConfig
392	fn leak_str(s: String) -> &'static str {
393	Box::leak(s.into_boxed_str())
394	}
395
396	// Build a path-based gateway config from a base URL
397	fn gw(base: &str) -> IpfsGatewayConfig {
398	IpfsGatewayConfig {
399	url: leak_str(base.to_string()),
400	gateway_type: IpfsGatewayType::Path,
401	bearer_token_env: None,
402	}
403	}
404
405	#[tokio::test]
406	async fn first_alternative_succeeds() {
407	// Arrange gateways A(original), B(success), C(not used)
408	let server_a = MockServer::start().await;
409	let server_b = MockServer::start().await;
410	let server_c = MockServer::start().await;
411
412	let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/ok";
413	let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);
414
415	// B returns 200
416	Mock::given(method("GET"))
417	.and(path(format!("/ipfs/{}", content_path)))
418	.respond_with(ResponseTemplate::new(200).set_body_string("B OK"))
419	.mount(&server_b)
420	.await;
421
422	let gateways = vec![
423	gw(&server_a.uri()),
424	gw(&server_b.uri()),
425	gw(&server_c.uri()),
426	];
427
428	// Act
429	let res =
430	super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;
431
432	// Assert
433	assert!(res.is_ok());
434	let body = res.unwrap().bytes().await.unwrap();
435	assert_eq!(body.as_ref(), b"B OK");
436	}
437
438	#[tokio::test]
439	async fn first_alternative_http_error_second_succeeds() {
440	// Arrange gateways A(original), B(500), C(200)
441	let server_a = MockServer::start().await;
442	let server_b = MockServer::start().await;
443	let server_c = MockServer::start().await;
444
445	let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/ok2";
446	let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);
447
448	// B returns 500
449	Mock::given(method("GET"))
450	.and(path(format!("/ipfs/{}", content_path)))
451	.respond_with(ResponseTemplate::new(500))
452	.mount(&server_b)
453	.await;
454
455	// C returns 200
456	Mock::given(method("GET"))
457	.and(path(format!("/ipfs/{}", content_path)))
458	.respond_with(ResponseTemplate::new(200).set_body_string("C OK"))
459	.mount(&server_c)
460	.await;
461
462	let gateways = vec![
463	gw(&server_a.uri()),
464	gw(&server_b.uri()),
465	gw(&server_c.uri()),
466	];
467
468	// Act
469	let res =
470	super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;
471
472	// Assert
473	assert!(res.is_ok());
474	let body = res.unwrap().bytes().await.unwrap();
475	assert_eq!(body.as_ref(), b"C OK");
476	}
477
478	#[tokio::test]
479	async fn first_alternative_non_http_error_second_succeeds() {
480	// Arrange gateways A(original), B(non-http error), C(200)
481	let server_a = MockServer::start().await;
482	let server_c = MockServer::start().await;
483
484	// Use an unroutable/closed port for B to simulate network error
485	let dead_gateway_base = "http://127.0.0.1:9"; // port 9 (discard) likely closed
486
487	let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/ok3";
488	let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);
489
490	// C returns 200
491	Mock::given(method("GET"))
492	.and(path(format!("/ipfs/{}", content_path)))
493	.respond_with(ResponseTemplate::new(200).set_body_string("C OK"))
494	.mount(&server_c)
495	.await;
496
497	let gateways = vec![
498	gw(&server_a.uri()),
499	gw(dead_gateway_base),
500	gw(&server_c.uri()),
501	];
502
503	// Act
504	let res =
505	super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;
506
507	// Assert
508	assert!(res.is_ok());
509	let body = res.unwrap().bytes().await.unwrap();
510	assert_eq!(body.as_ref(), b"C OK");
511	}
512
513	#[tokio::test]
514	async fn both_alternatives_fail() {
515	// Arrange gateways A(original), B(500), C(502)
516	let server_a = MockServer::start().await;
517	let server_b = MockServer::start().await;
518	let server_c = MockServer::start().await;
519
520	let content_path = "QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco/fail";
521	let original_url = format!("{}/ipfs/{}", server_a.uri(), content_path);
522
523	// B returns 500
524	Mock::given(method("GET"))
525	.and(path(format!("/ipfs/{}", content_path)))
526	.respond_with(ResponseTemplate::new(500))
527	.mount(&server_b)
528	.await;
529
530	// C returns 502
531	Mock::given(method("GET"))
532	.and(path(format!("/ipfs/{}", content_path)))
533	.respond_with(ResponseTemplate::new(502))
534	.mount(&server_c)
535	.await;
536
537	let gateways = vec![
538	gw(&server_a.uri()),
539	gw(&server_b.uri()),
540	gw(&server_c.uri()),
541	];
542
543	// Act
544	let res =
545	super::retry_with_gateways(&original_url, anyhow::anyhow!("orig err"), &gateways).await;
546
547	// Assert: error should reflect the last error (502)
548	assert!(res.is_err());
549	let err = res.err().unwrap().to_string();
550	assert!(err.contains("502"));
551	}
552
553	#[tokio::test]
554	async fn non_ipfs_url_returns_original_error_without_warning() {
555	// Test that non-IPFS URLs return the original error without logging IPFS gateway warnings
556	let non_ipfs_url =
557	"https://mirror.xyz/10/0xceda033195af537e16d203a4d7fbfe1c5f0eb843/render";
558	let original_error_msg = "HTTP error: status 429 Too Many Requests";
559	let original_error = anyhow::anyhow!(original_error_msg);
560	let gateways = vec![gw("https://ipfs.io")];
561
562	// Act
563	let res = super::retry_with_gateways(non_ipfs_url, original_error, &gateways).await;
564
565	// Assert: should return the original error without trying IPFS gateways
566	assert!(res.is_err());
567	let err = res.err().unwrap();
568	assert_eq!(err.to_string(), original_error_msg);
569	}
570	}
571
572	#[cfg(test)]
573	mod fetch_url_tests {
574	use wiremock::{
575	matchers::{header, method, path},
576	Mock, MockServer, ResponseTemplate,
577	};
578
579	#[tokio::test]
580	async fn test_fetch_url_without_bearer_token() {
581	let mock_server = MockServer::start().await;
582	let url = format!("{}/test", mock_server.uri());
583
584	Mock::given(method("GET"))
585	.and(path("/test"))
586	.and(header("User-Agent", crate::USER_AGENT))
587	.respond_with(ResponseTemplate::new(200).set_body_string("OK"))
588	.mount(&mock_server)
589	.await;
590
591	let result = super::fetch_url(&url, None).await;
592	assert!(result.is_ok());
593	let response = result.unwrap();
594	assert_eq!(response.status(), reqwest::StatusCode::OK);
595	}
596
597	#[tokio::test]
598	async fn test_fetch_url_with_bearer_token() {
599	let mock_server = MockServer::start().await;
600	let url = format!("{}/protected", mock_server.uri());
601	let token = "test-bearer-token-123";
602	let auth_header = format!("Bearer {}", token);
603
604	Mock::given(method("GET"))
605	.and(path("/protected"))
606	.and(header("User-Agent", crate::USER_AGENT))
607	.and(header("Authorization", auth_header.as_str()))
608	.respond_with(ResponseTemplate::new(200).set_body_string("Protected OK"))
609	.mount(&mock_server)
610	.await;
611
612	let result = super::fetch_url(&url, Some(token.to_string())).await;
613	assert!(result.is_ok());
614	let response = result.unwrap();
615	assert_eq!(response.status(), reqwest::StatusCode::OK);
616	}
617
618	#[tokio::test]
619	async fn test_fetch_url_with_empty_bearer_token() {
620	let mock_server = MockServer::start().await;
621	let url = format!("{}/test", mock_server.uri());
622
623	Mock::given(method("GET"))
624	.and(path("/test"))
625	.and(header("User-Agent", crate::USER_AGENT))
626	.respond_with(ResponseTemplate::new(200).set_body_string("OK"))
627	.mount(&mock_server)
628	.await;
629
630	// Empty string should not add Authorization header
631	let result = super::fetch_url(&url, Some("".to_string())).await;
632	assert!(result.is_ok());
633	let response = result.unwrap();
634	assert_eq!(response.status(), reqwest::StatusCode::OK);
635	}
636
637	#[tokio::test]
638	async fn test_fetch_url_network_error() {
639	// Use a non-existent URL to test network error handling
640	let url = "http://127.0.0.1:9/nonexistent";
641
642	let result = super::fetch_url(url, None).await;
643	assert!(result.is_err());
644	}
645	}

0xmichalis / nftbk / 19676380878

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous