euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-29 18:22:38 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

Retry request from reverseproxy retry options

Browse Source
This commit is contained in:
euzu
2026-01-22 13:30:59 +01:00
parent 7dbd94792d
commit ccdbf1dfce
25 changed files with 414 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/api_utils.rs
+49 -94
View File
@@ -7,10 +7,10 @@ use crate::api::model::{create_channel_unavailable_stream, create_custom_video_s
use crate::api::model::{tee_stream, UserSession};
use crate::api::model::{ProviderAllocation, ProviderConfig, ProviderStreamState, StreamDetails, StreamingStrategy};
use crate::auth::Fingerprint;
use crate::model::{ConfigInput, ResourceRetryConfig};
use crate::model::{ConfigInput};
use crate::model::{ConfigTarget, ProxyUserCredentials};
use crate::tools::lru_cache::LRUResourceCache;
use crate::utils::request::{content_type_from_ext, parse_range};
use crate::utils::request::{content_type_from_ext, parse_range, send_with_retry};
use crate::utils::{async_file_reader, async_file_writer, create_new_file_for_write, get_file_extension};
use crate::utils::{debug_if_enabled, trace_if_enabled};
use crate::utils::request;
@@ -25,7 +25,6 @@ use chrono::{DateTime, Utc};
use futures::{stream, StreamExt, TryStreamExt};
use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation};
use log::{debug, error, info, log_enabled, trace, warn};
use reqwest::header::RETRY_AFTER;
use serde::Serialize;
use shared::concat_string;
use shared::model::{Claims, InputFetchMethod, PlaylistEntry, PlaylistItemType, ProxyType, StreamChannel, TargetType, UserConnectionPermission, VirtualId, XtreamCluster};
@@ -39,7 +38,6 @@ use std::convert::Infallible;
use std::io::SeekFrom;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncSeekExt};
use tokio::sync::Mutex;
use tokio_util::io::ReaderStream;
@@ -1343,102 +1341,59 @@ async fn fetch_resource_with_retry(
) -> Option<axum::response::Response> {
let config = app_state.app_config.config.load();
let default_user_agent = config.default_user_agent.clone();
let (max_attempts, backoff_ms, backoff_multiplier) = config
.reverse_proxy
.as_ref()
.map_or_else(ResourceRetryConfig::get_default_retry_values, |rp| rp.resource_retry.get_retry_values());
drop(config);
let disabled_headers = app_state.get_disabled_headers();
for attempt in 0..max_attempts {
let client = request::get_client_request(
&app_state.http_client.load(),
input.map_or(InputFetchMethod::GET, |i| i.method),
input.map(|i| &i.headers),
url,
Some(req_headers),
disabled_headers.as_ref(),
default_user_agent.as_deref(),
let response = match send_with_retry(
&app_state.app_config,
url,
|| {
request::get_client_request(
&app_state.http_client.load(),
input.map_or(InputFetchMethod::GET, |i| i.method),
input.map(|i| &i.headers),
url,
Some(req_headers),
disabled_headers.as_ref(),
default_user_agent.as_deref(),
)
},
)
.await
{
Ok(response) => response,
Err(_) => return None,
};
let status = response.status();
if status.is_success() {
return Some(
build_resource_stream_response(app_state, resource_url, response).await,
);
match client.send().await {
Ok(response) => {
let status = response.status();
if status.is_success() {
return Some(
build_resource_stream_response(app_state, resource_url, response).await,
);
}
// Retry only for 408, 425, 429 and all 5xx statuses
let should_retry = status.is_server_error()
|| matches!(
status,
// reqwest::StatusCode::BAD_REQUEST // 400 is typically client error; retrying likely won't help and adds load.
reqwest::StatusCode::REQUEST_TIMEOUT
| reqwest::StatusCode::TOO_EARLY
| reqwest::StatusCode::TOO_MANY_REQUESTS
);
if attempt < max_attempts - 1 && should_retry {
let wait_dur = response
.headers()
.get(RETRY_AFTER)
.and_then(|h| h.to_str().ok())
.and_then(|s| s.parse::<u64>().ok())
.map_or_else(
|| {
let delay = calculate_retry_backoff(backoff_ms, backoff_multiplier, attempt);
Duration::from_millis(delay)
},
Duration::from_secs,
);
tokio::time::sleep(wait_dur).await;
continue;
}
// For non-retriable statuses or when attempts are exhausted, return upstream response including body
debug_if_enabled!(
"Failed to open resource got status {status} for {}",
sanitize_sensitive_info(resource_url)
);
let mut response_builder = axum::response::Response::builder().status(status);
for (key, value) in response.headers() {
response_builder = response_builder.header(key, value);
}
let stream = response
.bytes_stream()
.map_err(|err| StreamError::reqwest(&err));
return Some(try_unwrap_body!(
response_builder.body(axum::body::Body::from_stream(stream))
));
}
Err(err) => {
if attempt < max_attempts - 1 {
let delay = calculate_retry_backoff(backoff_ms, backoff_multiplier, attempt);
tokio::time::sleep(Duration::from_millis(delay)).await;
continue;
}
error!("Received failure from server {}: {err}", sanitize_sensitive_info(resource_url));
}
}
break;
}
None
// Non-retriable Status → Upstream Response incl. Body
debug_if_enabled!(
"Failed to open resource got status {status} for {}",
sanitize_sensitive_info(resource_url)
);
let mut response_builder = axum::response::Response::builder().status(status);
for (key, value) in response.headers() {
response_builder = response_builder.header(key, value);
}
let stream = response
.bytes_stream()
.map_err(|err| StreamError::reqwest(&err));
Some(try_unwrap_body!(
response_builder.body(axum::body::Body::from_stream(stream))
))
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss, clippy::cast_precision_loss)]
fn calculate_retry_backoff(base_delay_ms: u64, multiplier: f64, attempt: u32) -> u64 {
let base = base_delay_ms.max(1);
if multiplier <= 1.0 {
return base;
}
let delay = (base as f64) * multiplier.powi(i32::try_from(attempt).unwrap_or(i32::MAX));
if !delay.is_finite() || delay < 1.0 {
base
} else if delay >= u64::MAX as f64 {
u64::MAX
} else {
delay as u64
}
}
/// # Panics
pub async fn resource_response(
backend/src/api/endpoints/api_playlist_utils.rs
+1 -1
View File
@@ -65,7 +65,7 @@ pub(in crate::api::endpoints) async fn get_playlist_for_custom_provider(client:
Some(input) => {
let (result, errors) =
match input.input_type {
InputType::M3u | InputType::M3uBatch => m3u::download_m3u_playlist(client, &cfg, input).await,
InputType::M3u | InputType::M3uBatch => m3u::download_m3u_playlist(app_config, client, &cfg, input).await,
InputType::Xtream | InputType::XtreamBatch => {
let (pl, err, _) = xtream::download_xtream_playlist(app_config, client, input, Some(&[cluster])).await;
(pl, err)
backend/src/api/endpoints/download_api.rs
+1 -4
View File
@@ -85,10 +85,7 @@ async fn run_download_queue(cfg: &AppConfig, download_cfg: &VideoDownloadConfig,
if next_download.is_some() {
{ *download_queue.as_ref().active.write().await = next_download; }
let config = cfg.config.load();
let disabled_headers = config
.reverse_proxy
.as_ref()
.and_then(|r| r.disabled_header.clone());
let disabled_headers = cfg.get_disabled_headers();
let headers = request::get_request_headers(
Some(&download_cfg.headers),
None,
backend/src/api/endpoints/hls_api.rs
+1 -2
View File
@@ -133,13 +133,12 @@ pub(in crate::api) async fn handle_hls_stream_request(
);
let input_source = InputSource::from(input).with_url(request_url);
match request::download_text_content(
&app_state.app_config,
&app_state.http_client.load(),
disabled_headers.as_ref(),
&input_source,
Some(&headers),
None,
false,
default_user_agent.as_deref(),
)
.await
{
backend/src/api/endpoints/v1_api.rs
+1 -4
View File
@@ -90,16 +90,13 @@ async fn geoip_update(axum::extract::State(app_state): axum::extract::State<Arc<
method: InputFetchMethod::GET,
headers: HashMap::default(),
};
let disabled_headers = app_state.get_disabled_headers();
let default_user_agent = config.default_user_agent.clone();
return match download_text_content(
&app_state.app_config,
&app_state.http_client.load(),
disabled_headers.as_ref(),
&input_source,
None,
None,
false,
default_user_agent.as_deref(),
)
.await
{
backend/src/api/endpoints/v1_api_config.rs
+1 -4
View File
@@ -147,16 +147,13 @@ async fn config_batch_content(
// The url is changed at this point, we need the raw url for the batch file
if let Some(batch_url) = config_input.t_batch_url.as_ref() {
let input_source = InputSource::from(&*config_input).with_url(batch_url.to_owned());
let disabled_headers = app_state.get_disabled_headers();
let default_user_agent = app_state.app_config.config.load().default_user_agent.clone();
return match download_text_content(
&app_state.app_config,
&app_state.http_client.load(),
disabled_headers.as_ref(),
&input_source,
None,
None,
false,
default_user_agent.as_deref(),
)
.await
{
backend/src/api/endpoints/v1_api_playlist.rs
+3 -1
View File
@@ -66,10 +66,12 @@ async fn playlist_update(
let event_manager = Arc::clone(&app_state.event_manager);
let playlist_state = Arc::clone(&app_state.playlists);
let valid_targets = Arc::new(valid_targets);
let disabled_headers = app_state.get_disabled_headers();
tokio::spawn({
async move {
playlist::exec_processing(&http_client, app_config, valid_targets, Some(event_manager),
Some(playlist_state), Some(app_state.update_guard.clone())).await;
Some(playlist_state), Some(app_state.update_guard.clone()),
disabled_headers).await;
}
});
axum::http::StatusCode::ACCEPTED.into_response()
backend/src/api/endpoints/xtream_api.rs
+9 -12
View File
@@ -12,10 +12,10 @@ use crate::auth::Fingerprint;
use crate::model::{xtream_mapping_option_from_target_options, ConfigTarget};
use crate::model::{Config, ConfigInput};
use crate::model::{InputSource, ProxyUserCredentials};
use crate::repository::{get_target_id_mapping, user_get_bouquet_filter, xtream_get_collection_path, xtream_get_item_for_stream_id, xtream_load_rewrite_playlist};
use crate::repository::get_target_storage_path;
use crate::repository::storage_const;
use crate::repository::VirtualIdRecord;
use crate::repository::{storage_const};
use crate::repository::{get_target_id_mapping, user_get_bouquet_filter, xtream_get_collection_path, xtream_get_item_for_stream_id, xtream_load_rewrite_playlist};
use crate::utils::xtream::create_vod_info_from_item;
use crate::utils::{debug_if_enabled, trace_if_enabled};
use crate::utils::{request, xtream};
@@ -903,13 +903,13 @@ async fn xtream_get_short_epg(
None,
).await {
let config = &app_state.app_config.config.load();
if let (Some(epg_path), Some(channel_id)) = (get_epg_path_for_target(config, target), &pli.epg_channel_id) {
if let Ok(exists) = tokio::fs::try_exists(&epg_path).await {
if exists {
return serve_short_epg(app_state, epg_path.as_path(), user, target, Arc::clone(channel_id), stream_id).await;
}
if let (Some(epg_path), Some(channel_id)) = (get_epg_path_for_target(config, target), &pli.epg_channel_id) {
if let Ok(exists) = tokio::fs::try_exists(&epg_path).await {
if exists {
return serve_short_epg(app_state, epg_path.as_path(), user, target, Arc::clone(channel_id), stream_id).await;
}
}
}
if pli.provider_id > 0 {
let input_name = &pli.input_name;
@@ -934,15 +934,13 @@ async fn xtream_get_short_epg(
// TODO serve epg from own db
let input_source = InputSource::from(&*input).with_url(info_url);
let default_user_agent = config.default_user_agent.clone();
return match request::download_text_content(
&app_state.app_config,
&app_state.http_client.load(),
None,
&input_source,
None,
None,
false,
default_user_agent.as_deref(),
)
.await
{
@@ -1041,13 +1039,12 @@ async fn xtream_get_catchup_response(
pli.provider_id
)));
let input_source = InputSource::from(&*input).with_url(info_url);
let default_user_agent = app_state.app_config.config.load().default_user_agent.clone();
let content = try_result_bad_request!(
xtream::get_xtream_stream_info_content(
&app_state.app_config,
&app_state.http_client.load(),
&input_source,
false,
default_user_agent.as_deref(),
)
.await
);
backend/src/api/main_api.rs
+2 -1
View File
@@ -132,8 +132,9 @@ fn exec_update_on_boot(
let playlist_state = Arc::clone(&app_state.playlists);
let client = client.clone();
let update_guard = Some(app_state.update_guard.clone());
let disabled_headers = app_state.get_disabled_headers();
tokio::spawn(async move {
playlist::exec_processing(&client, app_config_clone, targets_clone, None, Some(playlist_state), update_guard).await;
playlist::exec_processing(&client, app_config_clone, targets_clone, None, Some(playlist_state), update_guard, disabled_headers).await;
});
}
}
backend/src/api/model/app_state.rs
+1 -7
View File
@@ -443,13 +443,7 @@ impl AppState {
}
pub fn get_disabled_headers(&self) -> Option<ReverseProxyDisabledHeaderConfig> {
self
.app_config
.config
.load()
.reverse_proxy
.as_ref()
.and_then(|r| r.disabled_header.clone())
self.app_config.get_disabled_headers()
}
}
backend/src/api/scheduler.rs
+2 -1
View File
@@ -62,9 +62,10 @@ async fn start_scheduler(client: reqwest::Client, expression: &str, app_state: A
let app_config = Arc::clone(&app_state.app_config);
let event_manager = Arc::clone(&app_state.event_manager);
let playlist_state = app_state.playlists.clone();
let disabled_headers = app_state.get_disabled_headers();
sync_panel_api_exp_dates_on_boot(&app_state).await;
exec_processing(&client, app_config, Arc::clone(&targets), Some(event_manager),
Some(playlist_state), Some(app_state.update_guard.clone())).await;
Some(playlist_state), Some(app_state.update_guard.clone()), disabled_headers).await;
}
() = cancel.cancelled() => {
break;