237899745/ystp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add runtime policy and observability
Some checks failed
CI / verify (push) Has been cancelled
Details

Browse Source
This commit is contained in:
237899745
2026-07-25 20:00:11 +08:00
parent f3c7a77a37
commit 64b1169e8c
32 changed files with 1236 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/api/admin.rs
View File

@@ -1298,6 +1298,7 @@ async fn update_stripe_config(
Some(admin_id),
)
.await?;
audit_config_action(&state, admin_id, "stripe", ip).await?;
Ok(Json(Envelope {
success: true,
@@ -1363,6 +1364,7 @@ async fn update_auth_config(
Some(admin_id),
)
.await?;
audit_config_action(&state, admin_id, "auth", ip).await?;
Ok(Json(Envelope {
success: true,
@@ -1520,6 +1522,7 @@ async fn update_mail_config(
Some(admin_id),
)
.await?;
audit_config_action(&state, admin_id, "mail", ip).await?;
Ok(Json(Envelope {
success: true,
@@ -1652,6 +1655,10 @@ async fn update_config(
if key.is_empty() {
return Err(AppError::new(ErrorCode::InvalidRequest, "key 不能为空"));
}
if key.len() > 100 {
return Err(AppError::new(ErrorCode::InvalidRequest, "key 过长"));
}
settings::validate_runtime_config_value(key, &req.value)?;
let row = sqlx::query_as::<_, ConfigRow>(
r#"
@@ -1673,12 +1680,38 @@ async fn update_config(
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "更新配置失败").with_source(err))?;
if matches!(key, "auth" | "features" | "rate_limits" | "file_limits") {
state.runtime_policy_cache.invalidate().await;
}
audit_config_action(&state, admin_id, key, ip).await?;
Ok(Json(Envelope {
success: true,
data: row,
}))
}
async fn audit_config_action(
state: &AppState,
admin_id: Uuid,
key: &str,
ip: IpAddr,
) -> Result<(), AppError> {
sqlx::query(
r#"
INSERT INTO audit_logs (user_id, action, resource_type, details, ip_address)
VALUES ($1, 'system_config_update', 'system_config', $2, $3::inet)
"#,
)
.bind(admin_id)
.bind(serde_json::json!({ "key": key }))
.bind(ip.to_string())
.execute(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "写入配置审计日志失败").with_source(err))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;

36
src/api/auth.rs
View File

@@ -81,11 +81,18 @@ async fn register(
Json(req): Json<RegisterRequest>,
) -> Result<Json<Envelope<RegisterResponse>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let policy = settings::runtime_policy(&state).await?;
if !policy.features.registration_enabled {
return Err(AppError::new(
ErrorCode::Forbidden,
"用户注册功能当前已关闭",
));
}
rate_limit::enforce(
&state,
"auth_register_ip",
&ip.to_string(),
10,
policy.rate_limits.register_ip_per_hour,
60 * 60,
"注册请求过于频繁，请稍后再试",
)
@@ -96,7 +103,7 @@ async fn register(
credentials::validate_password(&req.password)?;
let password_hash = credentials::hash_password(&req.password).await?;
let verification_required = settings::email_verification_required(&state).await?;
let verification_required = policy.auth.email_verification_required;
let verified_at = (!verification_required).then(Utc::now);
let user = sqlx::query_as::<_, UserRow>(
@@ -199,11 +206,12 @@ async fn login(
}
let ip = context::client_ip(&headers, addr.ip());
let policy = settings::runtime_policy(&state).await?;
rate_limit::enforce(
&state,
"auth_login_ip",
&ip.to_string(),
30,
policy.rate_limits.login_ip_per_5_minutes,
5 * 60,
"登录请求过于频繁，请稍后再试",
)
@@ -212,7 +220,7 @@ async fn login(
&state,
"auth_login_identity",
&identity.to_lowercase(),
10,
policy.rate_limits.login_identity_per_5_minutes,
5 * 60,
"该账号登录尝试过于频繁，请稍后再试",
)
@@ -269,7 +277,7 @@ async fn login(
if !credentials::verify_password(&req.password, &user.password_hash).await? {
return Err(AppError::new(ErrorCode::Unauthorized, "账号或密码错误"));
}
let verification_required = settings::email_verification_required(&state).await?;
let verification_required = policy.auth.email_verification_required;
let (token, expires_at) = auth::issue_jwt(
&state.config.jwt_secret,
@@ -305,8 +313,9 @@ async fn send_verification(
headers: HeaderMap,
) -> Result<Json<Envelope<MessageResponse>>, AppError> {
let claims = auth::require_jwt(&state.config.jwt_secret, &headers)?;
let policy = settings::runtime_policy(&state).await?;
if !settings::email_verification_required(&state).await? {
if !policy.auth.email_verification_required {
return Ok(Json(Envelope {
success: true,
data: MessageResponse {
@@ -319,7 +328,7 @@ async fn send_verification(
&state,
"auth_send_verification_user",
&claims.sub.to_string(),
1,
policy.rate_limits.verification_email_per_minute,
60,
"发送过于频繁，请稍后再试",
)
@@ -414,11 +423,12 @@ async fn verify_email(
}
let ip = context::client_ip(&headers, addr.ip());
let policy = settings::runtime_policy(&state).await?;
rate_limit::enforce(
&state,
"auth_verify_email_ip",
&ip.to_string(),
20,
policy.rate_limits.email_verify_ip_per_15_minutes,
15 * 60,
"验证请求过于频繁，请稍后再试",
)
@@ -484,11 +494,12 @@ async fn forgot_password(
credentials::validate_email(&req.email)?;
let ip = context::client_ip(&headers, addr.ip());
let policy = settings::runtime_policy(&state).await?;
rate_limit::enforce(
&state,
"auth_forgot_ip",
&ip.to_string(),
5,
policy.rate_limits.forgot_password_ip_per_15_minutes,
15 * 60,
"找回密码请求过于频繁，请稍后再试",
)
@@ -497,7 +508,7 @@ async fn forgot_password(
&state,
"auth_forgot_email",
&req.email.to_lowercase(),
3,
policy.rate_limits.forgot_password_email_per_15_minutes,
15 * 60,
"找回密码请求过于频繁，请稍后再试",
)
@@ -575,11 +586,12 @@ async fn reset_password(
credentials::validate_password(&req.new_password)?;
let ip = context::client_ip(&headers, addr.ip());
let policy = settings::runtime_policy(&state).await?;
rate_limit::enforce(
&state,
"auth_reset_ip",
&ip.to_string(),
10,
policy.rate_limits.password_reset_ip_per_15_minutes,
15 * 60,
"重置密码请求过于频繁，请稍后再试",
)
@@ -588,7 +600,7 @@ async fn reset_password(
&state,
"auth_reset_token",
&req.token,
5,
policy.rate_limits.password_reset_token_per_15_minutes,
15 * 60,
"该重置链接尝试次数过多，请重新申请",
)

1
src/api/compress.rs
View File

@@ -111,6 +111,7 @@ async fn compress_json(
let ip = context::client_ip(&headers, addr.ip());
let (jar, principal) = context::authenticate(&state, jar, &headers, ip).await?;
context::require_api_permission(&principal, &["compress"])?;
context::enforce_anonymous_upload_rate(&state, &principal, ip).await?;
let admission = prepare_single_admission(&state, &principal, ip, true).await?;
let mut req = parse_single_file_request(

79
src/api/context.rs
View File

@@ -1,5 +1,6 @@
use crate::auth;
use crate::error::{AppError, ErrorCode};
use crate::services::{rate_limit, settings};
use crate::state::AppState;
use axum::http::HeaderMap;
@@ -101,7 +102,8 @@ pub async fn authenticate(
return Ok((jar, principal));
}
if !state.config.allow_anonymous_upload {
let policy = settings::runtime_policy(state).await?;
if !policy.features.anonymous_upload_enabled {
return Err(AppError::new(ErrorCode::Unauthorized, "未登录"));
}
@@ -115,6 +117,39 @@ pub async fn authenticate(
Ok((jar, Principal::Anonymous { session_id }))
}
pub async fn enforce_anonymous_upload_rate(
state: &AppState,
principal: &Principal,
ip: IpAddr,
) -> Result<(), AppError> {
let Principal::Anonymous { session_id } = principal else {
return Ok(());
};
let limit = settings::runtime_policy(state)
.await?
.rate_limits
.anonymous_per_minute;
rate_limit::enforce(
state,
"anonymous_upload_session",
session_id,
limit,
60,
"匿名上传请求过于频繁，请稍后再试",
)
.await?;
rate_limit::enforce(
state,
"anonymous_upload_ip",
&ip.to_string(),
limit,
60,
"匿名上传请求过于频繁，请稍后再试",
)
.await
}
async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result<Option<Principal>, AppError> {
let auth_header = headers
.get(axum::http::header::AUTHORIZATION)
@@ -134,11 +169,13 @@ async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result<Option<Princip
is_active: bool,
email_verified_at: Option<DateTime<Utc>>,
token_version: i32,
rate_limit_override: Option<i32>,
}
let user = sqlx::query_as::<_, UserAuthRow>(
r#"
SELECT id, role::text AS role, is_active, email_verified_at, token_version
SELECT id, role::text AS role, is_active, email_verified_at, token_version,
rate_limit_override
FROM users
WHERE id = $1
"#,
@@ -159,13 +196,28 @@ async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result<Option<Princip
));
}
let verification_required =
crate::services::settings::email_verification_required(state).await?;
let policy = settings::runtime_policy(state).await?;
let request_limit = user
.rate_limit_override
.filter(|limit| *limit > 0)
.map(|limit| limit as u32)
.unwrap_or(policy.rate_limits.user_per_minute)
.clamp(1, 100_000);
rate_limit::enforce(
state,
"user",
&user.id.to_string(),
request_limit,
60,
"账号请求频率已超过限制",
)
.await?;
Ok(Some(Principal::User {
user_id: user.id,
role: user.role,
email_verified: user.email_verified_at.is_some() || !verification_required,
email_verified: user.email_verified_at.is_some()
|| !policy.auth.email_verification_required,
}))
}
@@ -185,6 +237,14 @@ async fn try_api_key(
return Ok(None);
};
let policy = settings::runtime_policy(state).await?;
if !policy.features.api_key_enabled {
return Err(AppError::new(
ErrorCode::Forbidden,
"API Key 功能当前已关闭",
));
}
let key_prefix = full_key
.get(0..16)
.ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "API Key 格式错误"))?;
@@ -234,11 +294,11 @@ async fn try_api_key(
return Err(AppError::new(ErrorCode::Unauthorized, "API Key 无效"));
}
crate::services::rate_limit::enforce(
rate_limit::enforce(
state,
"api_key",
&row.id.to_string(),
row.rate_limit.clamp(1, 100_000) as u32,
(row.rate_limit.clamp(1, 100_000) as u32).min(policy.rate_limits.api_key_per_minute),
60,
"API Key 请求频率已超过限制",
)
@@ -264,14 +324,11 @@ async fn try_api_key(
.execute(&state.db)
.await;
let verification_required =
crate::services::settings::email_verification_required(state).await?;
Ok(Some(Principal::ApiKey {
user_id: row.user_id,
api_key_id: row.id,
role: row.user_role,
email_verified: row.email_verified_at.is_some() || !verification_required,
email_verified: row.email_verified_at.is_some() || !policy.auth.email_verification_required,
permissions,
}))
}

155
src/api/metrics.rs Normal file
View File

@@ -0,0 +1,155 @@
use crate::services::metrics::{
self, CLUSTER_METRICS_KEY, DEAD_STREAM_KEY, QUEUE_GROUP_NAME, QUEUE_STREAM_KEY,
};
use crate::state::AppState;
use axum::extract::State;
use axum::http::header::{CACHE_CONTROL, CONTENT_TYPE};
use axum::response::IntoResponse;
use redis::streams::StreamPendingReply;
use redis::AsyncCommands;
use std::collections::HashMap;
use std::fmt::Write;
use std::time::Duration;
const SCRAPE_TIMEOUT: Duration = Duration::from_secs(2);
pub async fn metrics(State(state): State<AppState>) -> impl IntoResponse {
let database = tokio::time::timeout(
SCRAPE_TIMEOUT,
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM tasks WHERE status IN ('pending', 'processing')",
)
.fetch_one(&state.db),
);
let redis = tokio::time::timeout(SCRAPE_TIMEOUT, redis_queue_stats(state.redis.clone()));
let (database, redis) = tokio::join!(database, redis);
let (database_up, active_tasks) = match database {
Ok(Ok(value)) => (1, value),
_ => (0, 0),
};
let (redis_up, queue_length, pending, dead_length, cluster) = match redis {
Ok(Ok((queue_length, pending, dead_length, cluster))) => {
(1, queue_length, pending, dead_length, cluster)
}
_ => (0, 0, 0, 0, HashMap::new()),
};
let mut output = metrics::render();
output
.push_str("# HELP imageforge_dependency_up Whether a required dependency is reachable.\n");
output.push_str("# TYPE imageforge_dependency_up gauge\n");
let _ = writeln!(
output,
"imageforge_dependency_up{{dependency=\"database\"}} {database_up}"
);
let _ = writeln!(
output,
"imageforge_dependency_up{{dependency=\"redis\"}} {redis_up}"
);
output.push_str("# HELP imageforge_active_tasks Current pending or processing tasks.\n");
output.push_str("# TYPE imageforge_active_tasks gauge\n");
let _ = writeln!(output, "imageforge_active_tasks {active_tasks}");
output.push_str("# HELP imageforge_queue_messages Current Redis stream message counts.\n");
output.push_str("# TYPE imageforge_queue_messages gauge\n");
let _ = writeln!(
output,
"imageforge_queue_messages{{state=\"stream\"}} {queue_length}"
);
let _ = writeln!(
output,
"imageforge_queue_messages{{state=\"pending\"}} {pending}"
);
let _ = writeln!(
output,
"imageforge_queue_messages{{state=\"dead_letter\"}} {dead_length}"
);
if redis_up == 1 {
render_cluster_counters(&mut output, &cluster);
}
(
[
(CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8"),
(CACHE_CONTROL, "no-store"),
],
output,
)
}
async fn redis_queue_stats(
mut connection: redis::aio::ConnectionManager,
) -> Result<(i64, usize, i64, HashMap<String, i64>), redis::RedisError> {
let queue_length: i64 = connection.xlen(QUEUE_STREAM_KEY).await?;
let pending = match connection
.xpending::<_, _, StreamPendingReply>(QUEUE_STREAM_KEY, QUEUE_GROUP_NAME)
.await
{
Ok(reply) => reply.count(),
Err(err) if err.to_string().contains("NOGROUP") => 0,
Err(err) => return Err(err),
};
let dead_length: i64 = connection.xlen(DEAD_STREAM_KEY).await?;
let cluster: HashMap<String, i64> = connection.hgetall(CLUSTER_METRICS_KEY).await?;
Ok((queue_length, pending, dead_length, cluster))
}
fn render_cluster_counters(output: &mut String, counters: &HashMap<String, i64>) {
let value = |name: &str| counters.get(name).copied().unwrap_or(0).max(0);
let success = value("compression_success");
let failed = value("compression_failed");
let duration_seconds = value("compression_duration_micros") as f64 / 1_000_000.0;
output.push_str("# HELP imageforge_compressions_total Compression attempts by result.\n");
output.push_str("# TYPE imageforge_compressions_total counter\n");
let _ = writeln!(
output,
"imageforge_compressions_total{{result=\"success\"}} {success}"
);
let _ = writeln!(
output,
"imageforge_compressions_total{{result=\"failed\"}} {failed}"
);
output.push_str(
"# HELP imageforge_compression_bytes_total Image bytes processed by direction.\n",
);
output.push_str("# TYPE imageforge_compression_bytes_total counter\n");
let _ = writeln!(
output,
"imageforge_compression_bytes_total{{direction=\"input\"}} {}",
value("compression_bytes_in")
);
let _ = writeln!(
output,
"imageforge_compression_bytes_total{{direction=\"output\"}} {}",
value("compression_bytes_out")
);
output.push_str("# HELP imageforge_compression_duration_seconds Total compression time.\n");
output.push_str("# TYPE imageforge_compression_duration_seconds summary\n");
let _ = writeln!(
output,
"imageforge_compression_duration_seconds_sum {duration_seconds}"
);
let _ = writeln!(
output,
"imageforge_compression_duration_seconds_count {}",
success + failed
);
output.push_str(
"# HELP imageforge_storage_fallbacks_total S3 writes that fell back to local storage.\n",
);
output.push_str("# TYPE imageforge_storage_fallbacks_total counter\n");
let _ = writeln!(
output,
"imageforge_storage_fallbacks_total {}",
value("storage_fallbacks")
);
output.push_str("# HELP imageforge_dead_letters_total Jobs moved to the dead-letter stream.\n");
output.push_str("# TYPE imageforge_dead_letters_total counter\n");
let _ = writeln!(
output,
"imageforge_dead_letters_total {}",
value("dead_letters")
);
}

6
src/api/mod.rs
View File

@@ -7,8 +7,10 @@ mod context;
mod downloads;
mod envelope;
mod health;
mod metrics;
mod multipart;
mod redemption;
pub(crate) mod request_context;
mod response;
mod tasks;
mod user;
@@ -21,7 +23,6 @@ use axum::extract::DefaultBodyLimit;
use axum::Router;
use std::net::SocketAddr;
use tower_http::services::{ServeDir, ServeFile};
use tower_http::trace::TraceLayer;
pub async fn run(state: AppState) -> Result<(), AppError> {
let addr = format!("{}:{}", state.config.host, state.config.port);
@@ -36,10 +37,11 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
let app = Router::new()
.route("/health", axum::routing::get(health::health))
.route("/metrics", axum::routing::get(metrics::metrics))
.nest("/downloads", downloads::router())
.nest("/api/v1", v1)
.fallback_service(static_service)
.layer(TraceLayer::new_for_http())
.layer(axum::middleware::from_fn(request_context::middleware))
.with_state(state);
let listener = tokio::net::TcpListener::bind(&addr)

125
src/api/request_context.rs Normal file
View File

@@ -0,0 +1,125 @@
use crate::services::metrics;
use axum::extract::Request;
use axum::http::HeaderValue;
use axum::middleware::Next;
use axum::response::Response;
use std::time::Instant;
use tracing::Instrument;
use uuid::Uuid;
tokio::task_local! {
static REQUEST_ID: String;
}
pub(crate) fn current_request_id() -> Option<String> {
REQUEST_ID.try_with(Clone::clone).ok()
}
pub(crate) async fn middleware(mut request: Request, next: Next) -> Response {
let request_id = request
.headers()
.get("x-request-id")
.and_then(|value| value.to_str().ok())
.filter(|value| valid_request_id(value))
.map(ToOwned::to_owned)
.unwrap_or_else(new_request_id);
let request_id_header = HeaderValue::from_str(&request_id)
.unwrap_or_else(|_| HeaderValue::from_static("invalid-request-id"));
request
.headers_mut()
.insert("x-request-id", request_id_header.clone());
let method = request.method().clone();
let path = request.uri().path().to_string();
let started = Instant::now();
let span = tracing::info_span!(
"http_request",
request_id = %request_id,
method = %method,
path = %path,
);
let mut response = REQUEST_ID
.scope(request_id.clone(), next.run(request).instrument(span))
.await;
let status = response.status();
let elapsed = started.elapsed();
if path != "/metrics" {
metrics::record_http(method.as_str(), status.as_u16(), elapsed);
}
if path == "/health" || path == "/metrics" {
tracing::debug!(
request_id = %request_id,
method = %method,
path = %path,
status = status.as_u16(),
latency_ms = elapsed.as_millis(),
"HTTP request completed"
);
} else {
tracing::info!(
request_id = %request_id,
method = %method,
path = %path,
status = status.as_u16(),
latency_ms = elapsed.as_millis(),
"HTTP request completed"
);
}
let headers = response.headers_mut();
headers.insert("x-request-id", request_id_header);
headers.insert(
"x-content-type-options",
HeaderValue::from_static("nosniff"),
);
headers.insert("x-frame-options", HeaderValue::from_static("SAMEORIGIN"));
headers.insert(
"referrer-policy",
HeaderValue::from_static("strict-origin-when-cross-origin"),
);
headers.insert(
"permissions-policy",
HeaderValue::from_static("camera=(), microphone=(), geolocation=()"),
);
headers.insert(
"content-security-policy",
HeaderValue::from_static(
"default-src 'self'; base-uri 'self'; object-src 'none'; frame-ancestors 'self'; form-action 'self'; img-src 'self' data: blob: https:; font-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self' https:",
),
);
response
}
fn new_request_id() -> String {
format!("req_{}", Uuid::new_v4())
}
fn valid_request_id(value: &str) -> bool {
!value.is_empty()
&& value.len() <= 128
&& value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-'))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn accepts_safe_gateway_request_ids() {
assert!(valid_request_id("req_1234-abcd.trace:01"));
}
#[test]
fn rejects_control_characters_and_oversized_ids() {
assert!(!valid_request_id(""));
assert!(!valid_request_id("request id"));
assert!(!valid_request_id("request\nspoof"));
assert!(!valid_request_id(&"a".repeat(129)));
}
}

1
src/api/tasks.rs
View File

@@ -93,6 +93,7 @@ async fn create_batch_task(
let ip = context::client_ip(&headers, addr.ip());
let (jar, principal) = context::authenticate(&state, jar, &headers, ip).await?;
context::require_api_permission(&principal, &["compress", "batch_compress"])?;
context::enforce_anonymous_upload_rate(&state, &principal, ip).await?;
let admission = prepare_batch_admission(&state, &principal).await?;
let idempotency_key = headers

20
src/api/user.rs
View File

@@ -697,6 +697,16 @@ async fn create_api_key(
if !email_verified {
return Err(AppError::new(ErrorCode::EmailNotVerified, "请先验证邮箱"));
}
if !settings::runtime_policy(&state)
.await?
.features
.api_key_enabled
{
return Err(AppError::new(
ErrorCode::Forbidden,
"API Key 功能当前已关闭",
));
}
let billing = billing::get_user_billing(&state, user_id).await?;
if !billing.plan.feature_api_enabled {
@@ -796,6 +806,16 @@ async fn rotate_api_key(
if !email_verified {
return Err(AppError::new(ErrorCode::EmailNotVerified, "请先验证邮箱"));
}
if !settings::runtime_policy(&state)
.await?
.features
.api_key_enabled
{
return Err(AppError::new(
ErrorCode::Forbidden,
"API Key 功能当前已关闭",
));
}
let (full_key, key_prefix) = generate_api_key();
let key_hash = context::api_key_hash(&full_key, &state.config.api_key_pepper)?;

6
src/error.rs
View File

@@ -5,7 +5,6 @@ use axum::{
};
use serde::Serialize;
use std::fmt::{Display, Formatter};
use uuid::Uuid;
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
@@ -98,7 +97,8 @@ struct ErrorPayload {
impl IntoResponse for AppError {
fn into_response(self) -> axum::response::Response {
let request_id = format!("req_{}", Uuid::new_v4());
let request_id = crate::api::request_context::current_request_id()
.unwrap_or_else(|| format!("req_{}", uuid::Uuid::new_v4()));
let status = match self.code {
ErrorCode::InvalidRequest => StatusCode::BAD_REQUEST,
@@ -144,6 +144,8 @@ impl IntoResponse for AppError {
},
};
crate::services::metrics::record_error(self.code);
let mut response = (status, Json(body)).into_response();
if let Ok(value) = HeaderValue::from_str(&request_id) {
response.headers_mut().insert("x-request-id", value);

1
src/main.rs
View File

@@ -43,6 +43,7 @@ async fn main() -> Result<(), AppError> {
redis,
mailer: std::sync::Arc::new(mailer),
image_processing_semaphore,
runtime_policy_cache: crate::services::settings::RuntimePolicyCache::new(),
};
match state.config.role.as_str() {

26
src/services/compress.rs
View File

@@ -16,6 +16,7 @@ use img_parts::{Bytes as ImgBytes, DynImage, ImageEXIF, ImageICC};
use oxipng::StripChunks;
use rgb::FromSlice;
use std::io::Cursor;
use std::time::Instant;
const TARGET_MIN_LONG_EDGE: u32 = 640;
const TARGET_MIN_SCALE: f64 = 0.55;
@@ -287,7 +288,12 @@ pub async fn compress_image_bytes(
max_height: Option<u32>,
preserve_metadata: bool,
) -> Result<Vec<u8>, AppError> {
let max_image_pixels = state.config.max_image_pixels;
let started = Instant::now();
let bytes_in = input.len() as u64;
let max_image_pixels = crate::services::settings::runtime_policy(state)
.await?
.file_limits
.max_image_pixels;
let permit = state
.image_processing_semaphore
.clone()
@@ -297,7 +303,7 @@ pub async fn compress_image_bytes(
AppError::new(ErrorCode::Internal, "图片处理并发控制器已关闭").with_source(err)
})?;
tokio::task::spawn_blocking(move || {
let result = match tokio::task::spawn_blocking(move || {
let _permit = permit;
compress_image_bytes_sync(
input,
@@ -313,9 +319,19 @@ pub async fn compress_image_bytes(
)
})
.await
.map_err(|err| {
AppError::new(ErrorCode::CompressionFailed, "图片处理任务异常退出").with_source(err)
})?
{
Ok(result) => result,
Err(err) => Err(
AppError::new(ErrorCode::CompressionFailed, "图片处理任务异常退出").with_source(err),
),
};
crate::services::metrics::record_compression(
state,
started.elapsed(),
bytes_in,
result.as_ref().ok().map(|bytes| bytes.len() as u64),
);
result
}
#[allow(clippy::too_many_arguments)]

273
src/services/metrics.rs Normal file
View File

@@ -0,0 +1,273 @@
use crate::error::ErrorCode;
use crate::state::AppState;
use std::fmt::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;
use std::time::Duration;
pub const QUEUE_STREAM_KEY: &str = "stream:compress_jobs";
pub const QUEUE_GROUP_NAME: &str = "compress_workers";
pub const DEAD_STREAM_KEY: &str = "stream:compress_jobs:dead";
pub const CLUSTER_METRICS_KEY: &str = "metrics:imageforge";
const METHODS: [&str; 5] = ["GET", "POST", "PUT", "DELETE", "OTHER"];
const STATUS_CLASSES: [&str; 5] = ["2xx", "3xx", "4xx", "5xx", "other"];
const ERROR_CODES: [&str; 17] = [
"INVALID_REQUEST",
"INVALID_IMAGE",
"UNSUPPORTED_FORMAT",
"TOO_MANY_PIXELS",
"FILE_TOO_LARGE",
"INVALID_TOKEN",
"UNAUTHORIZED",
"FORBIDDEN",
"NOT_FOUND",
"IDEMPOTENCY_CONFLICT",
"RATE_LIMITED",
"QUOTA_EXCEEDED",
"EMAIL_NOT_VERIFIED",
"COMPRESSION_FAILED",
"STORAGE_UNAVAILABLE",
"MAIL_SEND_FAILED",
"INTERNAL",
];
const DURATION_BUCKETS: [f64; 9] = [0.01, 0.05, 0.1, 0.3, 1.0, 3.0, 10.0, 30.0, f64::INFINITY];
struct Histogram {
buckets: [AtomicU64; DURATION_BUCKETS.len()],
count: AtomicU64,
sum_micros: AtomicU64,
}
impl Histogram {
fn new() -> Self {
Self {
buckets: std::array::from_fn(|_| AtomicU64::new(0)),
count: AtomicU64::new(0),
sum_micros: AtomicU64::new(0),
}
}
fn observe(&self, duration: Duration) {
let seconds = duration.as_secs_f64();
for (index, upper_bound) in DURATION_BUCKETS.iter().enumerate() {
if seconds <= *upper_bound {
self.buckets[index].fetch_add(1, Ordering::Relaxed);
}
}
self.count.fetch_add(1, Ordering::Relaxed);
self.sum_micros.fetch_add(
duration.as_micros().min(u64::MAX as u128) as u64,
Ordering::Relaxed,
);
}
fn render(&self, output: &mut String, name: &str) {
for (index, upper_bound) in DURATION_BUCKETS.iter().enumerate() {
let label = if upper_bound.is_infinite() {
"+Inf".to_string()
} else {
upper_bound.to_string()
};
let _ = writeln!(
output,
"{name}_bucket{{le=\"{label}\"}} {}",
self.buckets[index].load(Ordering::Relaxed)
);
}
let _ = writeln!(
output,
"{name}_sum {}",
self.sum_micros.load(Ordering::Relaxed) as f64 / 1_000_000.0
);
let _ = writeln!(
output,
"{name}_count {}",
self.count.load(Ordering::Relaxed)
);
}
}
struct Metrics {
http_requests: [[AtomicU64; STATUS_CLASSES.len()]; METHODS.len()],
http_duration: Histogram,
errors: [AtomicU64; ERROR_CODES.len()],
}
impl Metrics {
fn new() -> Self {
Self {
http_requests: std::array::from_fn(|_| std::array::from_fn(|_| AtomicU64::new(0))),
http_duration: Histogram::new(),
errors: std::array::from_fn(|_| AtomicU64::new(0)),
}
}
}
fn registry() -> &'static Metrics {
static METRICS: OnceLock<Metrics> = OnceLock::new();
METRICS.get_or_init(Metrics::new)
}
pub fn record_http(method: &str, status: u16, duration: Duration) {
let method_index = match method {
"GET" => 0,
"POST" => 1,
"PUT" => 2,
"DELETE" => 3,
_ => 4,
};
let status_index = match status {
200..=299 => 0,
300..=399 => 1,
400..=499 => 2,
500..=599 => 3,
_ => 4,
};
let metrics = registry();
metrics.http_requests[method_index][status_index].fetch_add(1, Ordering::Relaxed);
metrics.http_duration.observe(duration);
}
pub fn record_error(code: ErrorCode) {
registry().errors[error_index(code)].fetch_add(1, Ordering::Relaxed);
}
pub fn record_compression(
state: &AppState,
duration: Duration,
bytes_in: u64,
bytes_out: Option<u64>,
) {
let mut increments = vec![
("compression_bytes_in", bytes_in),
(
"compression_duration_micros",
duration.as_micros().min(u64::MAX as u128) as u64,
),
];
if let Some(bytes_out) = bytes_out {
increments.push(("compression_success", 1));
increments.push(("compression_bytes_out", bytes_out));
} else {
increments.push(("compression_failed", 1));
}
persist_cluster_increments(state, increments);
}
pub fn record_storage_fallback(state: &AppState) {
persist_cluster_increments(state, vec![("storage_fallbacks", 1)]);
}
pub fn record_dead_letter(state: &AppState) {
persist_cluster_increments(state, vec![("dead_letters", 1)]);
}
pub fn render() -> String {
let metrics = registry();
let mut output = String::with_capacity(8 * 1024);
output.push_str(
"# HELP imageforge_http_requests_total HTTP requests handled by method and status class.\n",
);
output.push_str("# TYPE imageforge_http_requests_total counter\n");
for (method_index, method) in METHODS.iter().enumerate() {
for (status_index, status_class) in STATUS_CLASSES.iter().enumerate() {
let _ = writeln!(
output,
"imageforge_http_requests_total{{method=\"{method}\",status_class=\"{status_class}\"}} {}",
metrics.http_requests[method_index][status_index].load(Ordering::Relaxed)
);
}
}
output.push_str("# HELP imageforge_http_request_duration_seconds HTTP request duration.\n");
output.push_str("# TYPE imageforge_http_request_duration_seconds histogram\n");
metrics
.http_duration
.render(&mut output, "imageforge_http_request_duration_seconds");
output.push_str("# HELP imageforge_errors_total Application errors by code.\n");
output.push_str("# TYPE imageforge_errors_total counter\n");
for (index, code) in ERROR_CODES.iter().enumerate() {
let _ = writeln!(
output,
"imageforge_errors_total{{code=\"{code}\"}} {}",
metrics.errors[index].load(Ordering::Relaxed)
);
}
output
}
fn persist_cluster_increments(state: &AppState, increments: Vec<(&'static str, u64)>) {
let mut connection = state.redis.clone();
tokio::spawn(async move {
let mut pipeline = redis::pipe();
for (field, amount) in increments {
pipeline
.cmd("HINCRBY")
.arg(CLUSTER_METRICS_KEY)
.arg(field)
.arg(amount.min(i64::MAX as u64) as i64)
.ignore();
}
if let Err(err) = pipeline.query_async::<_, ()>(&mut connection).await {
tracing::debug!(error = %err, "failed to persist cluster metric");
}
});
}
fn error_index(code: ErrorCode) -> usize {
match code {
ErrorCode::InvalidRequest => 0,
ErrorCode::InvalidImage => 1,
ErrorCode::UnsupportedFormat => 2,
ErrorCode::TooManyPixels => 3,
ErrorCode::FileTooLarge => 4,
ErrorCode::InvalidToken => 5,
ErrorCode::Unauthorized => 6,
ErrorCode::Forbidden => 7,
ErrorCode::NotFound => 8,
ErrorCode::IdempotencyConflict => 9,
ErrorCode::RateLimited => 10,
ErrorCode::QuotaExceeded => 11,
ErrorCode::EmailNotVerified => 12,
ErrorCode::CompressionFailed => 13,
ErrorCode::StorageUnavailable => 14,
ErrorCode::MailSendFailed => 15,
ErrorCode::Internal => 16,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn error_codes_and_slots_stay_aligned() {
let codes = [
ErrorCode::InvalidRequest,
ErrorCode::InvalidImage,
ErrorCode::UnsupportedFormat,
ErrorCode::TooManyPixels,
ErrorCode::FileTooLarge,
ErrorCode::InvalidToken,
ErrorCode::Unauthorized,
ErrorCode::Forbidden,
ErrorCode::NotFound,
ErrorCode::IdempotencyConflict,
ErrorCode::RateLimited,
ErrorCode::QuotaExceeded,
ErrorCode::EmailNotVerified,
ErrorCode::CompressionFailed,
ErrorCode::StorageUnavailable,
ErrorCode::MailSendFailed,
ErrorCode::Internal,
];
for (index, code) in codes.into_iter().enumerate() {
assert_eq!(error_index(code), index);
assert_eq!(ERROR_CODES[index], code.as_str());
}
}
}

1
src/services/mod.rs
View File

@@ -5,6 +5,7 @@ pub mod credentials;
pub mod filename;
pub mod idempotency;
pub mod mail;
pub mod metrics;
pub mod quota;
pub mod rate_limit;
pub mod settings;

7
src/services/quota.rs
View File

@@ -259,7 +259,10 @@ pub async fn consume_anonymous_units(
let mut conn = state.redis.clone();
let limit = state.config.anon_daily_units as i64;
let limit = crate::services::settings::runtime_policy(state)
.await?
.rate_limits
.anonymous_units_per_day as i64;
let ttl_seconds = 48 * 60 * 60;
let inc = units as i64;
@@ -299,7 +302,7 @@ pub async fn consume_anonymous_units(
if new_value < 0 {
return Err(AppError::new(
ErrorCode::QuotaExceeded,
"匿名试用次数已用完（每日 10 次）",
format!("匿名试用次数已用完（每日 {limit} 次）"),
));
}

348
src/services/settings.rs
View File

@@ -9,6 +9,13 @@ use rand::RngCore;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use sqlx::FromRow;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::time::Instant;
const RUNTIME_POLICY_CACHE_TTL: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MailCustomSmtp {
@@ -38,6 +45,151 @@ fn default_true() -> bool {
true
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FeaturesConfigStored {
#[serde(default = "default_true")]
pub registration_enabled: bool,
#[serde(default = "default_true")]
pub api_key_enabled: bool,
#[serde(default = "default_true")]
pub anonymous_upload_enabled: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RateLimitsConfigStored {
#[serde(default = "default_anonymous_per_minute")]
pub anonymous_per_minute: u32,
#[serde(default = "default_anonymous_units_per_day")]
pub anonymous_units_per_day: u32,
#[serde(default = "default_user_per_minute")]
pub user_per_minute: u32,
#[serde(default = "default_api_key_per_minute")]
pub api_key_per_minute: u32,
#[serde(default = "default_login_ip_per_5_minutes")]
pub login_ip_per_5_minutes: u32,
#[serde(default = "default_login_identity_per_5_minutes")]
pub login_identity_per_5_minutes: u32,
#[serde(default = "default_register_ip_per_hour")]
pub register_ip_per_hour: u32,
#[serde(default = "default_verification_email_per_minute")]
pub verification_email_per_minute: u32,
#[serde(default = "default_email_verify_ip_per_15_minutes")]
pub email_verify_ip_per_15_minutes: u32,
#[serde(default = "default_forgot_password_ip_per_15_minutes")]
pub forgot_password_ip_per_15_minutes: u32,
#[serde(default = "default_forgot_password_email_per_15_minutes")]
pub forgot_password_email_per_15_minutes: u32,
#[serde(default = "default_password_reset_ip_per_15_minutes")]
pub password_reset_ip_per_15_minutes: u32,
#[serde(default = "default_password_reset_token_per_15_minutes")]
pub password_reset_token_per_15_minutes: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileLimitsConfigStored {
#[serde(default = "default_max_image_pixels")]
pub max_image_pixels: u64,
}
#[derive(Debug, Clone)]
pub struct RuntimePolicy {
pub auth: AuthConfigStored,
pub features: FeaturesConfigStored,
pub rate_limits: RateLimitsConfigStored,
pub file_limits: FileLimitsConfigStored,
}
#[derive(Debug, Clone)]
struct CachedRuntimePolicy {
loaded_at: Instant,
policy: RuntimePolicy,
}
#[derive(Debug, Clone, Default)]
pub struct RuntimePolicyCache {
inner: Arc<RwLock<Option<CachedRuntimePolicy>>>,
}
impl RuntimePolicyCache {
pub fn new() -> Self {
Self::default()
}
async fn get(&self) -> Option<RuntimePolicy> {
let cache = self.inner.read().await;
cache.as_ref().and_then(|cached| {
(cached.loaded_at.elapsed() < RUNTIME_POLICY_CACHE_TTL).then(|| cached.policy.clone())
})
}
async fn set(&self, policy: RuntimePolicy) {
*self.inner.write().await = Some(CachedRuntimePolicy {
loaded_at: Instant::now(),
policy,
});
}
pub async fn invalidate(&self) {
*self.inner.write().await = None;
}
}
fn default_anonymous_per_minute() -> u32 {
10
}
fn default_anonymous_units_per_day() -> u32 {
10
}
fn default_user_per_minute() -> u32 {
60
}
fn default_api_key_per_minute() -> u32 {
100
}
fn default_login_ip_per_5_minutes() -> u32 {
30
}
fn default_login_identity_per_5_minutes() -> u32 {
10
}
fn default_register_ip_per_hour() -> u32 {
10
}
fn default_verification_email_per_minute() -> u32 {
1
}
fn default_email_verify_ip_per_15_minutes() -> u32 {
20
}
fn default_forgot_password_ip_per_15_minutes() -> u32 {
5
}
fn default_forgot_password_email_per_15_minutes() -> u32 {
3
}
fn default_password_reset_ip_per_15_minutes() -> u32 {
10
}
fn default_password_reset_token_per_15_minutes() -> u32 {
5
}
fn default_max_image_pixels() -> u64 {
40_000_000
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StripeConfigStored {
pub secret_key_encrypted: Option<String>,
@@ -73,6 +225,144 @@ pub async fn load_system_config<T: DeserializeOwned>(
Ok(Some(parsed))
}
pub async fn runtime_policy(state: &AppState) -> Result<RuntimePolicy, AppError> {
if let Some(policy) = state.runtime_policy_cache.get().await {
return Ok(policy);
}
#[derive(Debug, FromRow)]
struct ConfigValueRow {
key: String,
value: serde_json::Value,
}
let rows = sqlx::query_as::<_, ConfigValueRow>(
r#"
SELECT key, value
FROM system_config
WHERE key = ANY($1)
"#,
)
.bind(["auth", "features", "rate_limits", "file_limits"])
.fetch_all(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询运行策略失败").with_source(err))?;
let mut auth = AuthConfigStored {
email_verification_required: true,
};
let mut features = FeaturesConfigStored {
registration_enabled: true,
api_key_enabled: true,
anonymous_upload_enabled: true,
};
let mut rate_limits = RateLimitsConfigStored {
anonymous_units_per_day: state.config.anon_daily_units,
..RateLimitsConfigStored::default()
};
let mut file_limits = FileLimitsConfigStored {
max_image_pixels: state.config.max_image_pixels,
};
for row in rows {
match row.key.as_str() {
"auth" => auth = parse_config_value(row.value, "认证配置")?,
"features" => features = parse_config_value(row.value, "功能开关")?,
"rate_limits" => rate_limits = parse_config_value(row.value, "限速配置")?,
"file_limits" => file_limits = parse_config_value(row.value, "文件限制")?,
_ => {}
}
}
features.anonymous_upload_enabled &= state.config.allow_anonymous_upload;
validate_rate_limits(&rate_limits)?;
validate_file_limits(&file_limits)?;
let policy = RuntimePolicy {
auth,
features,
rate_limits,
file_limits,
};
state.runtime_policy_cache.set(policy.clone()).await;
Ok(policy)
}
fn parse_config_value<T: DeserializeOwned>(
value: serde_json::Value,
label: &str,
) -> Result<T, AppError> {
serde_json::from_value(value).map_err(|err| {
AppError::new(ErrorCode::Internal, format!("{label}格式错误")).with_source(err)
})
}
pub fn validate_runtime_config_value(key: &str, value: &serde_json::Value) -> Result<(), AppError> {
match key {
"auth" => {
serde_json::from_value::<AuthConfigStored>(value.clone()).map_err(|err| {
AppError::new(ErrorCode::InvalidRequest, "认证配置格式错误").with_source(err)
})?;
}
"features" => {
serde_json::from_value::<FeaturesConfigStored>(value.clone()).map_err(|err| {
AppError::new(ErrorCode::InvalidRequest, "功能开关格式错误").with_source(err)
})?;
}
"rate_limits" => {
let config =
serde_json::from_value::<RateLimitsConfigStored>(value.clone()).map_err(|err| {
AppError::new(ErrorCode::InvalidRequest, "限速配置格式错误").with_source(err)
})?;
validate_rate_limits(&config)?;
}
"file_limits" => {
let config =
serde_json::from_value::<FileLimitsConfigStored>(value.clone()).map_err(|err| {
AppError::new(ErrorCode::InvalidRequest, "文件限制格式错误").with_source(err)
})?;
validate_file_limits(&config)?;
}
_ => {}
}
Ok(())
}
fn validate_rate_limits(config: &RateLimitsConfigStored) -> Result<(), AppError> {
let values = [
config.anonymous_per_minute,
config.anonymous_units_per_day,
config.user_per_minute,
config.api_key_per_minute,
config.login_ip_per_5_minutes,
config.login_identity_per_5_minutes,
config.register_ip_per_hour,
config.verification_email_per_minute,
config.email_verify_ip_per_15_minutes,
config.forgot_password_ip_per_15_minutes,
config.forgot_password_email_per_15_minutes,
config.password_reset_ip_per_15_minutes,
config.password_reset_token_per_15_minutes,
];
if values.iter().any(|value| !(1..=100_000).contains(value)) {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"限速值必须在 1 到 100000 之间",
));
}
Ok(())
}
fn validate_file_limits(config: &FileLimitsConfigStored) -> Result<(), AppError> {
if !(1_000_000..=200_000_000).contains(&config.max_image_pixels) {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"max_image_pixels 必须在 1000000 到 200000000 之间",
));
}
Ok(())
}
pub async fn upsert_system_config(
state: &AppState,
key: &str,
@@ -99,6 +389,10 @@ pub async fn upsert_system_config(
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "更新系统配置失败").with_source(err))?;
if matches!(key, "auth" | "features" | "rate_limits" | "file_limits") {
state.runtime_policy_cache.invalidate().await;
}
Ok(())
}
@@ -128,10 +422,10 @@ pub async fn load_mail_settings(state: &AppState) -> Result<Option<MailSettings>
}
pub async fn email_verification_required(state: &AppState) -> Result<bool, AppError> {
Ok(load_system_config::<AuthConfigStored>(state, "auth")
Ok(runtime_policy(state)
.await?
.map(|config| config.email_verification_required)
.unwrap_or(true))
.auth
.email_verification_required)
}
pub async fn load_stripe_secrets(state: &AppState) -> Result<Option<StripeSecrets>, AppError> {
@@ -243,6 +537,26 @@ pub async fn get_stripe_webhook_secret(state: &AppState) -> Result<String, AppEr
.ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "未配置 Stripe Webhook Secret"))
}
impl Default for RateLimitsConfigStored {
fn default() -> Self {
Self {
anonymous_per_minute: default_anonymous_per_minute(),
anonymous_units_per_day: default_anonymous_units_per_day(),
user_per_minute: default_user_per_minute(),
api_key_per_minute: default_api_key_per_minute(),
login_ip_per_5_minutes: default_login_ip_per_5_minutes(),
login_identity_per_5_minutes: default_login_identity_per_5_minutes(),
register_ip_per_hour: default_register_ip_per_hour(),
verification_email_per_minute: default_verification_email_per_minute(),
email_verify_ip_per_15_minutes: default_email_verify_ip_per_15_minutes(),
forgot_password_ip_per_15_minutes: default_forgot_password_ip_per_15_minutes(),
forgot_password_email_per_15_minutes: default_forgot_password_email_per_15_minutes(),
password_reset_ip_per_15_minutes: default_password_reset_ip_per_15_minutes(),
password_reset_token_per_15_minutes: default_password_reset_token_per_15_minutes(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -252,4 +566,32 @@ mod tests {
let config: AuthConfigStored = serde_json::from_value(serde_json::json!({})).unwrap();
assert!(config.email_verification_required);
}
#[test]
fn rate_limit_defaults_support_legacy_config_rows() {
let config: RateLimitsConfigStored = serde_json::from_value(serde_json::json!({
"anonymous_per_minute": 7,
"user_per_minute": 55
}))
.unwrap();
assert_eq!(config.anonymous_per_minute, 7);
assert_eq!(config.user_per_minute, 55);
assert_eq!(config.login_identity_per_5_minutes, 10);
assert_eq!(config.password_reset_token_per_15_minutes, 5);
}
#[test]
fn invalid_runtime_limits_are_rejected_before_persisting() {
let result = validate_runtime_config_value(
"rate_limits",
&serde_json::json!({ "anonymous_per_minute": 0 }),
);
assert!(result.is_err());
let result = validate_runtime_config_value(
"file_limits",
&serde_json::json!({ "max_image_pixels": 999_999 }),
);
assert!(result.is_err());
}
}

7
src/services/storage.rs
View File

@@ -161,7 +161,7 @@ where
if let Some(endpoint) = active_endpoint(state).await? {
match store_bytes_s3(state, &endpoint, key, bytes.clone(), content_type).await {
Ok(stored) => return Ok(stored),
Err(err) => log_local_fallback(&endpoint, key, &err),
Err(err) => log_local_fallback(state, &endpoint, key, &err),
}
}
@@ -233,7 +233,7 @@ pub async fn store_file(
if let Some(endpoint) = active_endpoint(state).await? {
match store_file_s3(state, &endpoint, key, path, content_type, metadata.len()).await {
Ok(stored) => return Ok(stored),
Err(err) => log_local_fallback(&endpoint, key, &err),
Err(err) => log_local_fallback(state, &endpoint, key, &err),
}
}
@@ -302,7 +302,8 @@ async fn store_file_local(
})
}
fn log_local_fallback(endpoint: &StorageEndpoint, key: &str, err: &AppError) {
fn log_local_fallback(state: &AppState, endpoint: &StorageEndpoint, key: &str, err: &AppError) {
crate::services::metrics::record_storage_fallback(state);
tracing::warn!(
storage_endpoint_id = %endpoint.id,
storage_endpoint = %endpoint.name,

2
src/state.rs
View File

@@ -1,5 +1,6 @@
use crate::config::Config;
use crate::services::mail::Mailer;
use crate::services::settings::RuntimePolicyCache;
#[derive(Clone)]
pub struct AppState {
@@ -8,4 +9,5 @@ pub struct AppState {
pub redis: redis::aio::ConnectionManager,
pub mailer: std::sync::Arc<Mailer>,
pub image_processing_semaphore: std::sync::Arc<tokio::sync::Semaphore>,
pub runtime_policy_cache: RuntimePolicyCache,
}

8
src/worker/mod.rs
View File

@@ -1,6 +1,7 @@
use crate::error::{AppError, ErrorCode};
use crate::services::billing;
use crate::services::compress;
use crate::services::metrics;
use crate::services::quota;
use crate::services::storage;
use crate::state::AppState;
@@ -15,9 +16,9 @@ use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use uuid::Uuid;
const STREAM_KEY: &str = "stream:compress_jobs";
const GROUP_NAME: &str = "compress_workers";
const DEAD_STREAM_KEY: &str = "stream:compress_jobs:dead";
const STREAM_KEY: &str = metrics::QUEUE_STREAM_KEY;
const GROUP_NAME: &str = metrics::QUEUE_GROUP_NAME;
const DEAD_STREAM_KEY: &str = metrics::DEAD_STREAM_KEY;
const MAX_DELIVERIES: usize = 3;
const STALE_MESSAGE_IDLE_MS: usize = 5 * 60 * 1000;
@@ -136,6 +137,7 @@ async fn handle_message(
write_dead_letter(conn, &msg.id, task_id, deliveries, &err).await?;
mark_task_dead_letter(state, task_id, &err.message).await?;
ack_message(conn, &msg.id).await?;
metrics::record_dead_letter(state);
tracing::error!(
task_id = %task_id,
deliveries,