didirus/AstralRinth
5
11
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases 15 Wiki Activity

feat(app-backend): key circuit breaker by URI path pattern (#6383)

Browse Source 
* feat(app-lib): split `FetchFence` on base URI path

* chore: fmt

* fix(app-lib): fix test timing
This commit is contained in:
François-Xavier Talbot
2026-06-16 14:49:06 -04:00
committed by GitHub
parent a3aeeac2c3
commit 3aaa2ef071
13 changed files with 155 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app-lib/src/util/fetch.rs
+97 -16
View File
@@ -10,7 +10,7 @@ use rand::Rng;
use reqwest::Method;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::collections::{HashMap, VecDeque};
use std::ffi::OsStr;
use std::path::{Path, PathBuf};
use std::sync::LazyLock;
@@ -50,20 +50,48 @@ pub struct IoSemaphore(pub Semaphore);
pub struct FetchSemaphore(pub Semaphore);
struct FetchFence {
inner: Mutex<FenceInner>,
inner: Mutex<HashMap<&'static str, FenceInner>>,
}
impl FetchFence {
pub fn is_blocked(&self) -> bool {
self.inner.lock().is_blocked()
pub fn is_blocked(&self, key: &'static str) -> bool {
self.inner
.lock()
.entry(key)
.or_insert_with(FenceInner::new)
.is_blocked()
}
pub fn record_ok(&self) {
self.inner.lock().record_ok()
pub fn record_ok(&self, key: &'static str) {
self.inner
.lock()
.entry(key)
.or_insert_with(FenceInner::new)
.record_ok()
}
pub fn record_fail(&self) {
self.inner.lock().record_fail()
pub fn record_fail(&self, key: &'static str) {
self.inner
.lock()
.entry(key)
.or_insert_with(FenceInner::new)
.record_fail()
}
pub fn latest_block_minutes(&self) -> u32 {
let now = Utc::now();
self.inner
.lock()
.values()
.filter_map(|fence| fence.block_until)
.filter(|until| *until > now)
.max()
.map(|until| {
let seconds = until.signed_duration_since(now).num_seconds();
(seconds.max(0) as u32).div_ceil(60).max(1)
})
.unwrap_or(1)
}
}
@@ -154,7 +182,7 @@ impl FenceInner {
static GLOBAL_FETCH_FENCE: LazyLock<FetchFence> =
LazyLock::new(|| FetchFence {
inner: Mutex::new(FenceInner::new()),
inner: Mutex::new(HashMap::new()),
});
fn reqwest_client_builder() -> reqwest::ClientBuilder {
@@ -184,6 +212,7 @@ pub async fn fetch(
url: &str,
sha1: Option<&str>,
download_meta: Option<&DownloadMeta>,
uri_path: Option<&'static str>,
semaphore: &FetchSemaphore,
exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
) -> crate::Result<Bytes> {
@@ -195,6 +224,7 @@ pub async fn fetch(
None,
download_meta,
None,
uri_path,
semaphore,
exec,
)
@@ -206,6 +236,7 @@ pub async fn fetch_with_client(
url: &str,
sha1: Option<&str>,
download_meta: Option<&DownloadMeta>,
uri_path: Option<&'static str>,
semaphore: &FetchSemaphore,
exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
client: &reqwest::Client,
@@ -218,6 +249,7 @@ pub async fn fetch_with_client(
None,
download_meta,
None,
uri_path,
semaphore,
exec,
client,
@@ -231,6 +263,7 @@ pub async fn fetch_json<T>(
url: &str,
sha1: Option<&str>,
json_body: Option<serde_json::Value>,
uri_path: Option<&'static str>,
semaphore: &FetchSemaphore,
exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
) -> crate::Result<T>
@@ -238,7 +271,8 @@ where
T: DeserializeOwned,
{
let result = fetch_advanced(
method, url, sha1, json_body, None, None, None, semaphore, exec,
method, url, sha1, json_body, None, None, None, uri_path, semaphore,
exec,
)
.await?;
let value = serde_json::from_slice(&result)?;
@@ -257,6 +291,7 @@ pub async fn fetch_advanced(
header: Option<(&str, &str)>,
download_meta: Option<&DownloadMeta>,
loading_bar: Option<(&LoadingBarId, f64)>,
uri_path: Option<&'static str>,
semaphore: &FetchSemaphore,
exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
) -> crate::Result<Bytes> {
@@ -268,6 +303,7 @@ pub async fn fetch_advanced(
header,
download_meta,
loading_bar,
uri_path,
semaphore,
exec,
&INSECURE_REQWEST_CLIENT,
@@ -286,6 +322,7 @@ pub async fn fetch_advanced_with_client(
header: Option<(&str, &str)>,
download_meta: Option<&DownloadMeta>,
loading_bar: Option<(&LoadingBarId, f64)>,
uri_path: Option<&'static str>,
semaphore: &FetchSemaphore,
exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
client: &reqwest::Client,
@@ -294,6 +331,7 @@ pub async fn fetch_advanced_with_client(
let is_api_url = url.starts_with(env!("MODRINTH_API_URL"))
|| url.starts_with(env!("MODRINTH_API_URL_V3"));
let fence_key = if is_api_url { uri_path } else { None };
let creds = if header
.as_ref()
@@ -309,8 +347,13 @@ pub async fn fetch_advanced_with_client(
.map(|m| (DOWNLOAD_META_HEADER.to_string(), m.to_header_value()));
for attempt in 1..=(FETCH_ATTEMPTS + 1) {
if is_api_url && GLOBAL_FETCH_FENCE.is_blocked() {
return Err(ErrorKind::ApiIsDownError.into());
if let Some(fence_key) = fence_key
&& GLOBAL_FETCH_FENCE.is_blocked(fence_key)
{
return Err(ErrorKind::ApiIsDownError(
GLOBAL_FETCH_FENCE.latest_block_minutes(),
)
.into());
}
let mut req = client.request(method.clone(), url);
@@ -336,8 +379,8 @@ pub async fn fetch_advanced_with_client(
match result {
Ok(resp) => {
if resp.status().is_server_error() {
if is_api_url {
GLOBAL_FETCH_FENCE.record_fail();
if let Some(fence_key) = fence_key {
GLOBAL_FETCH_FENCE.record_fail(fence_key);
}
if attempt <= FETCH_ATTEMPTS {
@@ -400,8 +443,8 @@ pub async fn fetch_advanced_with_client(
tracing::trace!("Done downloading URL {url}");
if is_api_url {
GLOBAL_FETCH_FENCE.record_ok();
if let Some(fence_key) = fence_key {
GLOBAL_FETCH_FENCE.record_ok(fence_key);
}
return Ok(bytes);
@@ -427,6 +470,7 @@ pub async fn fetch_mirrors(
mirrors: &[&str],
sha1: Option<&str>,
download_meta: Option<&DownloadMeta>,
uri_path: Option<&'static str>,
semaphore: &FetchSemaphore,
exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite> + Copy,
) -> crate::Result<Bytes> {
@@ -441,6 +485,7 @@ pub async fn fetch_mirrors(
mirror,
sha1,
download_meta,
uri_path,
semaphore,
exec,
&REQWEST_CLIENT,
@@ -620,6 +665,42 @@ mod tests {
assert!(fence.is_blocked());
}
#[test]
fn test_fetch_fence_keys_are_independent() {
let fence = FetchFence {
inner: Mutex::new(HashMap::new()),
};
for _ in 0..FenceInner::FAILURE_THRESHOLD {
fence.record_fail("/v3/version_file/:sha1/update");
}
assert!(fence.is_blocked("/v3/version_file/:sha1/update"));
assert!(!fence.is_blocked("/v3/project/:id"));
}
#[test]
fn test_fetch_fence_latest_block_minutes() {
let fence = FetchFence {
inner: Mutex::new(HashMap::new()),
};
{
let mut inner = fence.inner.lock();
inner.insert("/expired", FenceInner::new());
inner.get_mut("/expired").unwrap().block_until =
Some(Utc::now() - TimeDelta::minutes(1));
inner.insert("/short", FenceInner::new());
inner.get_mut("/short").unwrap().block_until =
Some(Utc::now() + TimeDelta::seconds(61));
inner.insert("/long", FenceInner::new());
inner.get_mut("/long").unwrap().block_until =
Some(Utc::now() + TimeDelta::seconds(140));
}
assert_eq!(fence.latest_block_minutes(), 3);
}
#[test]
fn test_fence_block_after_4_fails_with_oks() {
// Update tests if the FenceInner constants change