fabi/EventSnap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'perf/stability-2026-07-07'

Browse Source 
Performance & stability batch: config cache, streaming uploads, disk-snapshot
cache, export video streaming, single-query auth, SSE resync-on-lag, quota
fail-open, graceful shutdown, and tx-failure cleanup. 40 backend unit tests +
179 e2e passing.
This commit is contained in:
fabi
2026-07-07 20:26:52 +02:00
parent 4cdb3ae14a d6c91974eb
commit 683b1b6f65
18 changed files with 745 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
backend/src/auth/handlers.rs
View File

@@ -37,8 +37,8 @@ pub async fn join(
Json(body): Json<JoinRequest>, Json(body): Json<JoinRequest>,
) -> Result<(StatusCode, Json<JoinResponse>), AppError> { ) -> Result<(StatusCode, Json<JoinResponse>), AppError> {
let ip = client_ip(&headers, "unknown"); let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let join_rate_on = config::get_bool(&state.pool, "join_rate_enabled", true).await; let join_rate_on = config::get_bool(&state.config_cache, "join_rate_enabled", true).await;
if rate_limits_on && join_rate_on if rate_limits_on && join_rate_on
&& !state.rate_limiter.check(format!("join:{ip}"), 5, Duration::from_secs(60)) && !state.rate_limiter.check(format!("join:{ip}"), 5, Duration::from_secs(60))
{ {
@@ -134,8 +134,8 @@ pub async fn recover(
// every 15 minutes, indefinitely. 5 attempts per 15 minutes per (IP, name) // every 15 minutes, indefinitely. 5 attempts per 15 minutes per (IP, name)
// softens that into a real cost. // softens that into a real cost.
let ip = client_ip(&headers, "unknown"); let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let recover_rate_on = config::get_bool(&state.pool, "recover_rate_enabled", true).await; let recover_rate_on = config::get_bool(&state.config_cache, "recover_rate_enabled", true).await;
if rate_limits_on && recover_rate_on { if rate_limits_on && recover_rate_on {
let name_key = display_name.to_lowercase(); let name_key = display_name.to_lowercase();
if !state.rate_limiter.check( if !state.rate_limiter.check(
@@ -260,8 +260,8 @@ pub async fn admin_login(
// a long-running guess campaign. 5 attempts / minute / IP is plenty for // a long-running guess campaign. 5 attempts / minute / IP is plenty for
// honest typos. // honest typos.
let ip = client_ip(&headers, "unknown"); let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let admin_rate_on = config::get_bool(&state.pool, "admin_login_rate_enabled", true).await; let admin_rate_on = config::get_bool(&state.config_cache, "admin_login_rate_enabled", true).await;
if rate_limits_on && admin_rate_on if rate_limits_on && admin_rate_on
&& !state.rate_limiter.check( && !state.rate_limiter.check(
format!("admin_login:{ip}"), format!("admin_login:{ip}"),

28
backend/src/auth/middleware.rs
View File

@@ -37,34 +37,32 @@ impl FromRequestParts<AppState> for AuthUser {
.strip_prefix("Bearer ") .strip_prefix("Bearer ")
.ok_or_else(|| AppError::Unauthorized("Ungültiges Token-Format.".into()))?; .ok_or_else(|| AppError::Unauthorized("Ungültiges Token-Format.".into()))?;
let claims = jwt::verify_token(token, &state.config.jwt_secret) // Verify the JWT (signature + expiry). We deliberately do NOT trust its role/ban
// claims — the live user row below is authoritative — so the decoded claims
// themselves aren't needed beyond this check.
jwt::verify_token(token, &state.config.jwt_secret)
.map_err(|_| AppError::Unauthorized("Token ungültig oder abgelaufen.".into()))?; .map_err(|_| AppError::Unauthorized("Token ungültig oder abgelaufen.".into()))?;
let token_hash = jwt::hash_token(token); let token_hash = jwt::hash_token(token);
let session = Session::find_by_token_hash(&state.pool, &token_hash) // Single round-trip: resolve the session token to its *live* user row. A
// role/ban stored in the token would survive a demote/ban for the full session
// lifetime (up to 30d), so we always re-read the user (a demoted host loses host
// powers immediately). We do NOT reject banned users here — they retain read
// access by design; writes and host/admin actions enforce the ban downstream.
let user = Session::find_user_by_token_hash(&state.pool, &token_hash)
.await .await
.map_err(|e| AppError::Internal(e.into()))? .map_err(|e| AppError::Internal(e.into()))?
.ok_or_else(|| AppError::Unauthorized("Sitzung nicht gefunden oder abgelaufen.".into()))?; .ok_or_else(|| AppError::Unauthorized("Sitzung nicht gefunden oder abgelaufen.".into()))?;
// Trust *live* DB state, not the JWT: a role/ban stored in the token would
// survive a demote/ban for the full session lifetime (up to 30d). Re-read
// the user row so a demoted host loses host powers immediately. We do NOT
// reject banned users here — they retain read access by design; writes and
// host/admin actions enforce the ban downstream.
let user = crate::models::user::User::find_by_id(&state.pool, claims.sub)
.await
.map_err(|e| AppError::Internal(e.into()))?
.ok_or_else(|| AppError::Unauthorized("Benutzer nicht gefunden.".into()))?;
// Update last_seen_at in the background (fire-and-forget). Failures are // Update last_seen_at in the background (fire-and-forget). Failures are
// non-fatal but worth surfacing — silent swallowing hides DB connection // non-fatal but worth surfacing — silent swallowing hides DB connection
// pressure that would otherwise be the first symptom of a real problem. // pressure that would otherwise be the first symptom of a real problem.
let pool = state.pool.clone(); let pool = state.pool.clone();
let session_id = session.id; let touch_hash = token_hash.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = Session::touch(&pool, session_id).await { if let Err(e) = Session::touch_by_token_hash(&pool, &touch_hash).await {
tracing::warn!(error = ?e, session_id = %session_id, "session touch failed"); tracing::warn!(error = ?e, "session touch failed");
} }
}); });

35
backend/src/handlers/admin.rs
View File

@@ -5,7 +5,6 @@ use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode}; use axum::http::{HeaderMap, StatusCode};
use axum::Json; use axum::Json;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use sysinfo::System;
use crate::auth::middleware::RequireAdmin; use crate::auth::middleware::RequireAdmin;
use crate::error::AppError; use crate::error::AppError;
@@ -68,23 +67,12 @@ pub async fn get_stats(
.fetch_one(&state.pool) .fetch_one(&state.pool)
.await?; .await?;
// Disk usage via sysinfo // Disk usage from the shared cache (unknown mount → zeros, same as before).
let mut sys = System::new(); let (disk_total, disk_free) = state
sys.refresh_all(); .disk_cache
.snapshot(&state.config.media_path)
let media_path = state.config.media_path.to_string_lossy().to_string(); .map(|d| (d.total, d.free))
let (disk_total, disk_free) = sysinfo::Disks::new_with_refreshed_list() .unwrap_or((0, 0));
.iter()
.find(|d| media_path.starts_with(d.mount_point().to_string_lossy().as_ref()))
.map(|d| (d.total_space(), d.available_space()))
.unwrap_or_else(|| {
// Fall back to the root disk
sysinfo::Disks::new_with_refreshed_list()
.iter()
.find(|d| d.mount_point().to_string_lossy() == "/")
.map(|d| (d.total_space(), d.available_space()))
.unwrap_or((0, 0))
});
let disk_used = disk_total.saturating_sub(disk_free); let disk_used = disk_total.saturating_sub(disk_free);
@@ -216,6 +204,11 @@ pub async fn patch_config(
} }
tx.commit().await?; tx.commit().await?;
// The config cache must reflect this write on the very next read (tests PATCH then
// immediately assert the new value takes effect). Invalidate synchronously here —
// the TTL is only a backstop and must not be relied on for correctness.
state.config_cache.invalidate();
// Notify all clients that a publicly-readable config value changed so their stores // Notify all clients that a publicly-readable config value changed so their stores
// (e.g. the privacy note in My Account) refresh without a manual reload. // (e.g. the privacy note in My Account) refresh without a manual reload.
if privacy_note_changed { if privacy_note_changed {
@@ -408,13 +401,13 @@ pub async fn export_status(
/// switch + per-endpoint switch + numeric value, all stored in `config` and read on /// switch + per-endpoint switch + numeric value, all stored in `config` and read on
/// each request. /// each request.
async fn enforce_export_rate(state: &AppState, headers: &HeaderMap) -> Result<(), AppError> { async fn enforce_export_rate(state: &AppState, headers: &HeaderMap) -> Result<(), AppError> {
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let export_rate_on = config::get_bool(&state.pool, "export_rate_enabled", true).await; let export_rate_on = config::get_bool(&state.config_cache, "export_rate_enabled", true).await;
if !(rate_limits_on && export_rate_on) { if !(rate_limits_on && export_rate_on) {
return Ok(()); return Ok(());
} }
let ip = client_ip(headers, "unknown"); let ip = client_ip(headers, "unknown");
let limit = config::get_usize(&state.pool, "export_rate_per_day", 3).await; let limit = config::get_usize(&state.config_cache, "export_rate_per_day", 3).await;
if !state if !state
.rate_limiter .rate_limiter
.check(format!("export:{ip}"), limit, Duration::from_secs(86400)) .check(format!("export:{ip}"), limit, Duration::from_secs(86400))

6
backend/src/handlers/feed.rs
View File

@@ -62,10 +62,10 @@ pub async fn feed(
Query(q): Query<FeedQuery>, Query(q): Query<FeedQuery>,
) -> Result<Json<FeedResponse>, AppError> { ) -> Result<Json<FeedResponse>, AppError> {
let ip = client_ip(&headers, "unknown"); let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let feed_rate_on = config::get_bool(&state.pool, "feed_rate_enabled", true).await; let feed_rate_on = config::get_bool(&state.config_cache, "feed_rate_enabled", true).await;
if rate_limits_on && feed_rate_on { if rate_limits_on && feed_rate_on {
let rate_limit = config::get_usize(&state.pool, "feed_rate_per_min", 60).await; let rate_limit = config::get_usize(&state.config_cache, "feed_rate_per_min", 60).await;
if !state if !state
.rate_limiter .rate_limiter
.check(format!("feed:{ip}"), rate_limit, Duration::from_secs(60)) .check(format!("feed:{ip}"), rate_limit, Duration::from_secs(60))

6
backend/src/handlers/me.rs
View File

@@ -65,9 +65,9 @@ pub async fn get_context(
.await? .await?
.ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?; .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;
let privacy_note = config::get_str(&state.pool, "privacy_note", "").await; let privacy_note = config::get_str(&state.config_cache, "privacy_note", "").await;
let quota_enabled = config::get_bool(&state.pool, "quota_enabled", true).await; let quota_enabled = config::get_bool(&state.config_cache, "quota_enabled", true).await;
let storage_quota_enabled = config::get_bool(&state.pool, "storage_quota_enabled", true).await; let storage_quota_enabled = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await;
Ok(Json(MeContextDto { Ok(Json(MeContextDto {
user_id: user.id, user_id: user.id,

10
backend/src/handlers/sse.rs
View File

@@ -6,6 +6,7 @@ use axum::response::sse::{Event, KeepAlive, Sse};
use axum::Json; use axum::Json;
use futures::stream::Stream; use futures::stream::Stream;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tokio_stream::wrappers::BroadcastStream; use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::StreamExt; use tokio_stream::StreamExt;
@@ -62,7 +63,14 @@ pub async fn stream(
Ok(sse_event) => Some(Ok(Event::default() Ok(sse_event) => Some(Ok(Event::default()
.event(sse_event.event_type) .event(sse_event.event_type)
.data(sse_event.data))), .data(sse_event.data))),
Err(_) => None, // A consumer that falls behind the broadcast buffer would otherwise silently
// lose events, leaving its feed permanently stale. Instead of dropping the
// gap, tell the client to resync — the frontend responds by running a full
// feed-delta fetch (which reconciles both new uploads and deletions).
Err(BroadcastStreamRecvError::Lagged(n)) => {
tracing::warn!("SSE consumer lagged, dropped {n} event(s); emitting resync");
Some(Ok(Event::default().event("resync").data(n.to_string())))
}
}); });
Ok(Sse::new(stream).keep_alive( Ok(Sse::new(stream).keep_alive(

5
backend/src/handlers/test_admin.rs
View File

@@ -80,6 +80,11 @@ pub async fn truncate_all(
// counters don't leak into the next one. // counters don't leak into the next one.
state.rate_limiter.clear(); state.rate_limiter.clear();
// The reseed above wrote the `config` table directly (bypassing patch_config), so
// the cache must be invalidated too — otherwise the first request after a truncate
// could serve the previous test's toggles.
state.config_cache.invalidate();
Ok(StatusCode::NO_CONTENT) Ok(StatusCode::NO_CONTENT)
} }

308
backend/src/handlers/upload.rs
View File

@@ -43,10 +43,10 @@ pub async fn upload(
mut multipart: Multipart, mut multipart: Multipart,
) -> Result<(StatusCode, Json<UploadDto>), AppError> { ) -> Result<(StatusCode, Json<UploadDto>), AppError> {
// Rate limit: N uploads per hour per user. Gated by master + per-endpoint toggles. // Rate limit: N uploads per hour per user. Gated by master + per-endpoint toggles.
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let upload_rate_on = config::get_bool(&state.pool, "upload_rate_enabled", true).await; let upload_rate_on = config::get_bool(&state.config_cache, "upload_rate_enabled", true).await;
if rate_limits_on && upload_rate_on { if rate_limits_on && upload_rate_on {
let upload_rate = config::get_i64(&state.pool, "upload_rate_per_hour", 10).await as usize; let upload_rate = config::get_i64(&state.config_cache, "upload_rate_per_hour", 10).await as usize;
if let Err(retry_after_secs) = state.rate_limiter.check_with_retry( if let Err(retry_after_secs) = state.rate_limiter.check_with_retry(
format!("upload:{}", auth.user_id), format!("upload:{}", auth.user_id),
upload_rate, upload_rate,
@@ -79,57 +79,85 @@ pub async fn upload(
} }
// Read config limits from DB // Read config limits from DB
let max_image_mb: i64 = config::get_i64(&state.pool, "max_image_size_mb", 20).await; let max_image_mb: i64 = config::get_i64(&state.config_cache, "max_image_size_mb", 20).await;
let max_video_mb: i64 = config::get_i64(&state.pool, "max_video_size_mb", 500).await; let max_video_mb: i64 = config::get_i64(&state.config_cache, "max_video_size_mb", 500).await;
let mut file_data: Option<Vec<u8>> = None; // The uploaded file is streamed straight to a temp file on disk (never buffered
// whole in memory — a 500 MB video used to cost 500 MB of RAM per concurrent
// upload). We only keep the first ≤512 bytes in memory for magic-byte sniffing.
// On success the temp file is renamed into place under its detected extension.
let upload_id = Uuid::new_v4();
let event_slug = &state.config.event_slug;
let originals_dir = state.config.media_path.join(format!("originals/{event_slug}"));
let temp_abs = originals_dir.join(format!("{upload_id}.tmp"));
let mut streamed: Option<(i64, Vec<u8>)> = None; // (size, head bytes for sniffing)
let mut caption: Option<String> = None; let mut caption: Option<String> = None;
let mut hashtags_csv: Option<String> = None; let mut hashtags_csv: Option<String> = None;
while let Some(field) = multipart.next_field().await.map_err(|e| AppError::BadRequest(e.to_string()))? { // Wrap the multipart read so any error after the temp file is created still cleans
let name = field.name().unwrap_or_default().to_string(); // it up (a mid-stream parse failure must not leave a stray `.tmp` on disk).
match name.as_str() { let parse_result: Result<(), AppError> = async {
"file" => { while let Some(field) = multipart
// Note: the client-declared filename and Content-Type do NOT determine .next_field()
// the stored MIME/extension — those come from the file's magic bytes .await
// below. The declared type is used only to pick a memory cap so an .map_err(|e| AppError::BadRequest(e.to_string()))?
// oversized body can't be fully buffered before the size check. A {
// mislabelled type only makes the cap *stricter* (safe); the let name = field.name().unwrap_or_default().to_string();
// authoritative per-class check still runs on the detected type. match name.as_str() {
let declared = field.content_type().unwrap_or("").to_string(); "file" => {
let cap_bytes = if declared.starts_with("video/") { // The client-declared Content-Type does NOT determine the stored
(max_video_mb * 1024 * 1024) as usize // MIME/extension — those come from the file's magic bytes below. The
} else if declared.starts_with("image/") { // declared type only picks the streaming cap so an oversized body is
(max_image_mb * 1024 * 1024) as usize // aborted early; a mislabelled type only makes the cap *stricter*
} else { // (safe), and the authoritative per-class check still runs on the
(max_image_mb.max(max_video_mb) * 1024 * 1024) as usize // detected type.
}; let declared = field.content_type().unwrap_or("").to_string();
file_data = Some(read_field_capped(field, cap_bytes).await?); let cap_bytes = if declared.starts_with("video/") {
(max_video_mb * 1024 * 1024) as usize
} else if declared.starts_with("image/") {
(max_image_mb * 1024 * 1024) as usize
} else {
(max_image_mb.max(max_video_mb) * 1024 * 1024) as usize
};
tokio::fs::create_dir_all(&originals_dir)
.await
.map_err(|e| AppError::Internal(e.into()))?;
streamed = Some(stream_field_to_file(field, &temp_abs, cap_bytes).await?);
}
"caption" => {
caption =
Some(field.text().await.map_err(|e| AppError::BadRequest(e.to_string()))?);
}
"hashtags" => {
hashtags_csv =
Some(field.text().await.map_err(|e| AppError::BadRequest(e.to_string()))?);
}
_ => {}
} }
"caption" => {
caption = Some(
field.text().await
.map_err(|e| AppError::BadRequest(e.to_string()))?,
);
}
"hashtags" => {
hashtags_csv = Some(
field.text().await
.map_err(|e| AppError::BadRequest(e.to_string()))?,
);
}
_ => {}
} }
Ok(())
}
.await;
if let Err(e) = parse_result {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(e);
} }
let data = file_data.ok_or_else(|| AppError::BadRequest("Keine Datei hochgeladen.".into()))?; // From here on the temp file may exist; every validation failure removes it before
let size = data.len() as i64; // returning so a rejected upload never leaves bytes behind.
let (size, head) = match streamed {
Some(s) => s,
None => return Err(AppError::BadRequest("Keine Datei hochgeladen.".into())),
};
// Validate caption length. Counted in chars (code points) to match the // Validate caption length. Counted in chars (code points) to match the
// "Zeichen" wording in the error message — `.len()` would be bytes and // "Zeichen" wording in the error message — `.len()` would be bytes and
// reject perfectly valid German/emoji captions early. // reject perfectly valid German/emoji captions early.
if let Some(ref cap) = caption { if let Some(ref cap) = caption {
if cap.chars().count() > MAX_CAPTION_LENGTH { if cap.chars().count() > MAX_CAPTION_LENGTH {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::BadRequest(format!( return Err(AppError::BadRequest(format!(
"Beschreibung ist zu lang. Maximum: {} Zeichen.", "Beschreibung ist zu lang. Maximum: {} Zeichen.",
MAX_CAPTION_LENGTH MAX_CAPTION_LENGTH
@@ -142,26 +170,38 @@ pub async fn upload(
// those are rejected outright — closing the stored-XSS vector. Both the MIME // those are rejected outright — closing the stored-XSS vector. Both the MIME
// we persist and the on-disk extension come from the detected type, never from // we persist and the on-disk extension come from the detected type, never from
// client-supplied values. // client-supplied values.
let kind = infer::get(&data) let kind = match infer::get(&head) {
.ok_or_else(|| AppError::BadRequest("Dateityp nicht erkannt oder nicht unterstützt.".into()))?; Some(k) => k,
let (mime, ext) = ALLOWED_MEDIA None => {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::BadRequest(
"Dateityp nicht erkannt oder nicht unterstützt.".into(),
));
}
};
let (mime, ext) = match ALLOWED_MEDIA
.iter() .iter()
.find(|(allowed, _)| *allowed == kind.mime_type()) .find(|(allowed, _)| *allowed == kind.mime_type())
.map(|(m, e)| ((*m).to_string(), *e)) .map(|(m, e)| ((*m).to_string(), *e))
.ok_or_else(|| { {
AppError::BadRequest(format!( Some(v) => v,
None => {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::BadRequest(format!(
"Dateityp wird nicht unterstützt: {}.", "Dateityp wird nicht unterstützt: {}.",
kind.mime_type() kind.mime_type()
)) )));
})?; }
};
// Validate file size // Validate file size against the authoritative per-detected-class limit.
let max_bytes = if mime.starts_with("video/") { let max_bytes = if mime.starts_with("video/") {
max_video_mb * 1024 * 1024 max_video_mb * 1024 * 1024
} else { } else {
max_image_mb * 1024 * 1024 max_image_mb * 1024 * 1024
}; };
if size > max_bytes { if size > max_bytes {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::BadRequest(format!( return Err(AppError::BadRequest(format!(
"Datei ist zu groß. Maximum: {} MB.", "Datei ist zu groß. Maximum: {} MB.",
max_bytes / (1024 * 1024) max_bytes / (1024 * 1024)
@@ -171,13 +211,14 @@ pub async fn upload(
// Per-user storage quota — dynamic formula based on available disk space and the // Per-user storage quota — dynamic formula based on available disk space and the
// number of active uploaders. Gated by master + per-area toggles so the admin can // number of active uploaders. Gated by master + per-area toggles so the admin can
// disable it on trusted instances. // disable it on trusted instances.
let quota_on = config::get_bool(&state.pool, "quota_enabled", true).await; let quota_on = config::get_bool(&state.config_cache, "quota_enabled", true).await;
let storage_quota_on = config::get_bool(&state.pool, "storage_quota_enabled", true).await; let storage_quota_on = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await;
if quota_on && storage_quota_on { if quota_on && storage_quota_on {
let estimate = compute_storage_quota(&state).await; let estimate = compute_storage_quota(&state).await;
if let Some(limit) = estimate.limit_bytes { if let Some(limit) = estimate.limit_bytes {
let prospective_total = user.total_upload_bytes.saturating_add(size); let prospective_total = user.total_upload_bytes.saturating_add(size);
if prospective_total > limit { if prospective_total > limit {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::TooManyRequests( return Err(AppError::TooManyRequests(
"Du hast dein Upload-Limit für dieses Event erreicht.".into(), "Du hast dein Upload-Limit für dieses Event erreicht.".into(),
None, None,
@@ -186,16 +227,13 @@ pub async fn upload(
} }
} }
let upload_id = Uuid::new_v4(); // All checks passed — atomically move the temp file to its final, extension-correct
let event_slug = &state.config.event_slug; // path (same directory, so the rename is cheap and atomic).
let relative_path = format!("originals/{event_slug}/{upload_id}.{ext}"); let relative_path = format!("originals/{event_slug}/{upload_id}.{ext}");
let absolute_path = state.config.media_path.join(&relative_path); let absolute_path = state.config.media_path.join(&relative_path);
tokio::fs::rename(&temp_abs, &absolute_path)
// Ensure directory exists and write file .await
if let Some(parent) = absolute_path.parent() { .map_err(|e| AppError::Internal(e.into()))?;
tokio::fs::create_dir_all(parent).await.map_err(|e| AppError::Internal(e.into()))?;
}
tokio::fs::write(&absolute_path, &data).await.map_err(|e| AppError::Internal(e.into()))?;
// Process hashtags from caption and explicit CSV // Process hashtags from caption and explicit CSV
let mut tags: Vec<String> = Vec::new(); let mut tags: Vec<String> = Vec::new();
@@ -216,27 +254,41 @@ pub async fn upload(
// Quota accounting, the upload row, and its hashtag links must be atomic: a // Quota accounting, the upload row, and its hashtag links must be atomic: a
// crash between the bytes increment and the insert would permanently charge // crash between the bytes increment and the insert would permanently charge
// bytes with no row to reclaim them (silent quota erosion / spurious lockout). // bytes with no row to reclaim them (silent quota erosion / spurious lockout).
let mut tx = state.pool.begin().await?; let tx_result: Result<Upload, AppError> = async {
sqlx::query("UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1") let mut tx = state.pool.begin().await?;
.bind(auth.user_id) sqlx::query("UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1")
.bind(size) .bind(auth.user_id)
.execute(&mut *tx) .bind(size)
.execute(&mut *tx)
.await?;
let upload = Upload::create(
&mut *tx,
auth.event_id,
auth.user_id,
&relative_path,
&mime,
size,
caption.as_deref(),
)
.await?; .await?;
let upload = Upload::create( for tag in &tags {
&mut *tx, let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?;
auth.event_id, Hashtag::link_to_upload(&mut *tx, upload.id, h.id).await?;
auth.user_id, }
&relative_path, tx.commit().await?;
&mime, Ok(upload)
size,
caption.as_deref(),
)
.await?;
for tag in &tags {
let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?;
Hashtag::link_to_upload(&mut *tx, upload.id, h.id).await?;
} }
tx.commit().await?; .await;
// The file is already on disk at `absolute_path`. If the transaction failed, no DB
// row will ever reference it, so remove it now rather than orphan bytes on disk.
let upload = match tx_result {
Ok(u) => u,
Err(e) => {
let _ = tokio::fs::remove_file(&absolute_path).await;
return Err(e);
}
};
// Spawn compression task // Spawn compression task
state state
@@ -339,30 +391,68 @@ pub async fn delete_upload(
Ok(StatusCode::NO_CONTENT) Ok(StatusCode::NO_CONTENT)
} }
/// Read a multipart field into memory, aborting with a 400 the moment it exceeds /// Number of leading bytes retained in memory for magic-byte (`infer`) sniffing. Every
/// `max_bytes`. Without this the whole field is buffered (up to the HTTP body cap) /// allowed type's signature sits well within this; 512 is comfortably generous.
/// before the post-read size check runs, so a request claiming to be a tiny image const HEAD_SNIFF_BYTES: usize = 512;
/// could still force hundreds of MB of allocation. Streaming with an early abort
/// bounds peak memory to roughly the applicable per-class limit. /// Stream a multipart field straight to `dest`, aborting with a 400 the moment it
async fn read_field_capped( /// exceeds `max_bytes`. Only the first [`HEAD_SNIFF_BYTES`] bytes are kept in memory
/// (for type detection); the rest goes chunk-by-chunk to disk, so peak memory is a
/// single chunk rather than the whole file. Returns `(total_size, head_bytes)`. On any
/// error the partial temp file is removed so no stray `.tmp` is left behind.
async fn stream_field_to_file(
mut field: axum::extract::multipart::Field<'_>, mut field: axum::extract::multipart::Field<'_>,
dest: &std::path::Path,
max_bytes: usize, max_bytes: usize,
) -> Result<Vec<u8>, AppError> { ) -> Result<(i64, Vec<u8>), AppError> {
let mut buf: Vec<u8> = Vec::new(); use tokio::io::AsyncWriteExt;
while let Some(chunk) = field
.chunk() let mut file = tokio::fs::File::create(dest)
.await .await
.map_err(|e| AppError::BadRequest(format!("Datei konnte nicht gelesen werden: {e}")))? .map_err(|e| AppError::Internal(e.into()))?;
{ let mut total: usize = 0;
if buf.len().saturating_add(chunk.len()) > max_bytes { let mut head: Vec<u8> = Vec::with_capacity(HEAD_SNIFF_BYTES);
loop {
let chunk = match field.chunk().await {
Ok(Some(c)) => c,
Ok(None) => break,
Err(e) => {
let _ = file.shutdown().await;
let _ = tokio::fs::remove_file(dest).await;
return Err(AppError::BadRequest(format!(
"Datei konnte nicht gelesen werden: {e}"
)));
}
};
total = total.saturating_add(chunk.len());
if total > max_bytes {
let _ = file.shutdown().await;
let _ = tokio::fs::remove_file(dest).await;
return Err(AppError::BadRequest(format!( return Err(AppError::BadRequest(format!(
"Datei ist zu groß. Maximum: {} MB.", "Datei ist zu groß. Maximum: {} MB.",
max_bytes / (1024 * 1024) max_bytes / (1024 * 1024)
))); )));
} }
buf.extend_from_slice(&chunk);
if head.len() < HEAD_SNIFF_BYTES {
let need = HEAD_SNIFF_BYTES - head.len();
head.extend_from_slice(&chunk[..need.min(chunk.len())]);
}
if let Err(e) = file.write_all(&chunk).await {
let _ = tokio::fs::remove_file(dest).await;
return Err(AppError::Internal(e.into()));
}
} }
Ok(buf)
if let Err(e) = file.flush().await {
let _ = tokio::fs::remove_file(dest).await;
return Err(AppError::Internal(e.into()));
}
Ok((total as i64, head))
} }
/// Drain a multipart body so the HTTP connection stays clean when returning an early error. /// Drain a multipart body so the HTTP connection stays clean when returning an early error.
@@ -396,9 +486,9 @@ fn quota_limit_bytes(free_disk: i64, tolerance: f64, active_uploaders: i64) -> i
/// None` whenever the storage quota is currently disabled — callers should skip the /// None` whenever the storage quota is currently disabled — callers should skip the
/// check (upload handler) or hide the UI (quota endpoint). /// check (upload handler) or hide the UI (quota endpoint).
pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate { pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate {
let quota_on = config::get_bool(&state.pool, "quota_enabled", true).await; let quota_on = config::get_bool(&state.config_cache, "quota_enabled", true).await;
let storage_quota_on = config::get_bool(&state.pool, "storage_quota_enabled", true).await; let storage_quota_on = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await;
let tolerance = config::get_f64(&state.pool, "quota_tolerance", 0.75).await; let tolerance = config::get_f64(&state.config_cache, "quota_tolerance", 0.75).await;
let (active_count,): (i64,) = sqlx::query_as( let (active_count,): (i64,) = sqlx::query_as(
"SELECT COUNT(DISTINCT user_id) FROM upload WHERE deleted_at IS NULL", "SELECT COUNT(DISTINCT user_id) FROM upload WHERE deleted_at IS NULL",
@@ -408,21 +498,23 @@ pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate {
.unwrap_or((0,)); .unwrap_or((0,));
let active = active_count.max(1); let active = active_count.max(1);
let media_path = state.config.media_path.to_string_lossy().to_string(); // Cached disk reading. `None` means we couldn't resolve the media filesystem.
let free_disk = sysinfo::Disks::new_with_refreshed_list() let disk = state.disk_cache.snapshot(&state.config.media_path);
.iter() let free_disk = disk.map(|d| d.free as i64).unwrap_or(0);
.find(|d| media_path.starts_with(d.mount_point().to_string_lossy().as_ref()))
.map(|d| d.available_space())
.unwrap_or_else(|| {
sysinfo::Disks::new_with_refreshed_list()
.iter()
.find(|d| d.mount_point().to_string_lossy() == "/")
.map(|d| d.available_space())
.unwrap_or(0)
}) as i64;
let limit_bytes = if quota_on && storage_quota_on { let limit_bytes = if quota_on && storage_quota_on {
Some(quota_limit_bytes(free_disk, tolerance, active)) match disk {
Some(d) => Some(quota_limit_bytes(d.free as i64, tolerance, active)),
// Fail OPEN, not closed: if the disk can't be read we don't know the real
// free space, and enforcing a 0-byte limit would reject every upload with a
// spurious "quota reached". Skip enforcement this round and warn instead.
None => {
tracing::warn!(
"disk snapshot unavailable; skipping storage-quota enforcement this round"
);
None
}
}
} else { } else {
None None
}; };

64
backend/src/main.rs
View File

@@ -161,7 +161,69 @@ async fn main() -> Result<()> {
let listener = tokio::net::TcpListener::bind(("0.0.0.0", config.app_port)).await?; let listener = tokio::net::TcpListener::bind(("0.0.0.0", config.app_port)).await?;
tracing::info!("listening on {}", listener.local_addr()?); tracing::info!("listening on {}", listener.local_addr()?);
axum::serve(listener, router).await?; axum::serve(listener, router)
.with_graceful_shutdown(shutdown_signal())
.await?;
Ok(()) Ok(())
} }
/// Hard cap on how long we wait for in-flight connections to drain after a shutdown
/// signal. Uploads (streamed to disk) finish in well under this; the cap exists because
/// long-lived SSE streams never end on their own and would otherwise keep the graceful
/// drain — and thus the process — pending until the orchestrator force-kills it.
const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(10);
/// Resolves on SIGINT (Ctrl-C) or SIGTERM (container stop / deploy). Letting
/// `axum::serve` drain in-flight requests on this signal means a redeploy no longer
/// truncates uploads mid-flight; any background compression/export half-states that a
/// hard kill would leave are already reconciled by `startup_recovery` on the next boot.
///
/// Once the signal fires we also arm a detached backstop that force-exits after
/// [`SHUTDOWN_GRACE`]. Without it, open SSE streams (which have no natural end) would
/// hold the graceful drain open indefinitely; the backstop bounds shutdown regardless
/// of the orchestrator's own kill timeout. If the drain completes first, `main` returns
/// and the process exits before the timer ever fires.
async fn shutdown_signal() {
let ctrl_c = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(unix)]
let terminate = async {
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(mut sig) => {
sig.recv().await;
}
Err(e) => {
tracing::warn!(error = ?e, "failed to install SIGTERM handler");
// Never resolve — fall back to ctrl_c only.
std::future::pending::<()>().await;
}
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {}
_ = terminate => {}
}
tracing::info!(
"shutdown signal received, draining in-flight requests (max {}s)",
SHUTDOWN_GRACE.as_secs()
);
// Backstop: if the graceful drain is still blocked after the grace window (almost
// always because SSE streams are still open), exit anyway so deploys aren't stalled.
tokio::spawn(async {
tokio::time::sleep(SHUTDOWN_GRACE).await;
tracing::warn!(
"graceful drain exceeded {}s (likely open SSE streams); forcing exit",
SHUTDOWN_GRACE.as_secs()
);
std::process::exit(0);
});
}

30
backend/src/models/session.rs
View File

@@ -43,9 +43,33 @@ impl Session {
.await .await
} }
pub async fn touch(pool: &PgPool, id: Uuid) -> Result<(), sqlx::Error> { /// Resolve a session token straight to its live user row in one round-trip.
sqlx::query("UPDATE session SET last_seen_at = NOW() WHERE id = $1") ///
.bind(id) /// The auth extractor runs on every authenticated request and used to do two
/// sequential queries (session lookup, then user lookup); this collapses them into
/// a single `session JOIN "user"`. The `expires_at` guard mirrors
/// [`Self::find_by_token_hash`], and the user row is read live (role/ban are never
/// trusted from the JWT). Returns `None` when the session is missing/expired.
pub async fn find_user_by_token_hash(
pool: &PgPool,
token_hash: &str,
) -> Result<Option<crate::models::user::User>, sqlx::Error> {
sqlx::query_as::<_, crate::models::user::User>(
"SELECT u.*
FROM session s
JOIN \"user\" u ON u.id = s.user_id
WHERE s.token_hash = $1 AND s.expires_at > NOW()",
)
.bind(token_hash)
.fetch_optional(pool)
.await
}
/// Update `last_seen_at`, keyed by the token hash so the auth extractor can touch a
/// session without a separate query to fetch its id first.
pub async fn touch_by_token_hash(pool: &PgPool, token_hash: &str) -> Result<(), sqlx::Error> {
sqlx::query("UPDATE session SET last_seen_at = NOW() WHERE token_hash = $1")
.bind(token_hash)
.execute(pool) .execute(pool)
.await?; .await?;
Ok(()) Ok(())

123
backend/src/services/config.rs
View File

@@ -1,46 +1,129 @@
//! Reads of the runtime-tunable `config` table. //! Reads of the runtime-tunable `config` table, fronted by an in-memory cache.
//! //!
//! Each handler used to keep a small local copy of these helpers; consolidating them //! Each handler used to keep a small local copy of these helpers; consolidating them
//! here means one place to add a parser, one place to mock for tests, and one place to //! here means one place to add a parser, one place to mock for tests, and one place to
//! find when a key changes. New keys do not require code changes — they're picked up //! find when a key changes. New keys do not require code changes — they're picked up
//! the next time someone calls `get_*`. //! the next time the cache reloads.
//!
//! ## Why a cache
//!
//! The `config` table is effectively static during an event, yet it was the busiest
//! query in the system: every request re-read each key with its own `SELECT`
//! (an upload touched it ~8 times). Against the small connection pool that was the
//! throughput ceiling. [`ConfigCache`] loads the whole table once and serves reads
//! from memory.
//!
//! ## Consistency contract
//!
//! Correctness comes from **synchronous invalidation on every write**, not from the
//! TTL. The two runtime write paths — the admin `PATCH /admin/config` handler and the
//! test-mode truncate/reseed — both call [`ConfigCache::invalidate`] after committing,
//! so the *next* read reloads from the DB and sees the new value immediately. The
//! [`RELOAD_TTL`] is only a safety net for out-of-band changes (e.g. a migration or a
//! manual DB edit); it is deliberately short but never the primary mechanism.
//! //!
//! Values are read with a default fallback so the app still starts if a key is missing //! Values are read with a default fallback so the app still starts if a key is missing
//! (e.g. during a migration window). Production seeds keys via migrations 005 and 009. //! (e.g. during a migration window). Production seeds keys via migrations 005 and 009.
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
use sqlx::PgPool; use sqlx::PgPool;
async fn fetch_raw(pool: &PgPool, key: &str) -> Option<String> { /// How long a loaded snapshot is trusted before the next read reloads it. This is a
sqlx::query_as::<_, (String,)>("SELECT value FROM config WHERE key = $1") /// backstop for out-of-band DB changes only — every in-process write invalidates the
.bind(key) /// cache synchronously, so tests that PATCH-then-assert never depend on this expiring.
.fetch_optional(pool) const RELOAD_TTL: Duration = Duration::from_secs(30);
.await
.ok() struct Snapshot {
.flatten() values: HashMap<String, String>,
.map(|(v,)| v) loaded_at: Instant,
} }
pub async fn get_str(pool: &PgPool, key: &str, default: &str) -> String { /// In-memory cache of the entire `config` table. Cheap to `clone` (shares the pool and
fetch_raw(pool, key).await.unwrap_or_else(|| default.to_string()) /// the `Arc`), so it lives in `AppState` and every handler reads through it.
#[derive(Clone)]
pub struct ConfigCache {
pool: PgPool,
inner: Arc<RwLock<Option<Snapshot>>>,
} }
pub async fn get_i64(pool: &PgPool, key: &str, default: i64) -> i64 { impl ConfigCache {
fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default) pub fn new(pool: PgPool) -> Self {
Self {
pool,
inner: Arc::new(RwLock::new(None)),
}
}
/// Drop the cached snapshot so the next read reloads the whole table from the DB.
/// Call this after any write to the `config` table (admin PATCH, test reseed).
pub fn invalidate(&self) {
*self.inner.write().unwrap() = None;
}
/// Return the fresh snapshot if one is loaded and still within [`RELOAD_TTL`].
fn fresh_snapshot(&self) -> Option<HashMap<String, String>> {
let guard = self.inner.read().unwrap();
match guard.as_ref() {
Some(snap) if snap.loaded_at.elapsed() < RELOAD_TTL => Some(snap.values.clone()),
_ => None,
}
}
/// Read one key, loading the whole table into the cache on a miss/expiry. On a DB
/// error we return `None` (callers fall back to their default) without poisoning
/// the cache.
async fn get_raw(&self, key: &str) -> Option<String> {
if let Some(values) = self.fresh_snapshot() {
return values.get(key).cloned();
}
// Cache miss or stale — reload the entire table in one query.
let rows: Vec<(String, String)> =
match sqlx::query_as::<_, (String, String)>("SELECT key, value FROM config")
.fetch_all(&self.pool)
.await
{
Ok(rows) => rows,
Err(e) => {
tracing::warn!(error = ?e, "config reload failed; using defaults for this read");
return None;
}
};
let values: HashMap<String, String> = rows.into_iter().collect();
let result = values.get(key).cloned();
*self.inner.write().unwrap() = Some(Snapshot {
values,
loaded_at: Instant::now(),
});
result
}
} }
pub async fn get_usize(pool: &PgPool, key: &str, default: usize) -> usize { pub async fn get_str(cache: &ConfigCache, key: &str, default: &str) -> String {
fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default) cache.get_raw(key).await.unwrap_or_else(|| default.to_string())
} }
pub async fn get_f64(pool: &PgPool, key: &str, default: f64) -> f64 { pub async fn get_i64(cache: &ConfigCache, key: &str, default: i64) -> i64 {
fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default) cache.get_raw(key).await.and_then(|v| v.parse().ok()).unwrap_or(default)
}
pub async fn get_usize(cache: &ConfigCache, key: &str, default: usize) -> usize {
cache.get_raw(key).await.and_then(|v| v.parse().ok()).unwrap_or(default)
}
pub async fn get_f64(cache: &ConfigCache, key: &str, default: f64) -> f64 {
cache.get_raw(key).await.and_then(|v| v.parse().ok()).unwrap_or(default)
} }
/// Parses common truthy spellings used by both the migration seeds and the admin form. /// Parses common truthy spellings used by both the migration seeds and the admin form.
/// Accepts `true/false`, `1/0`, `yes/no`, `on/off` — case-insensitive. Anything else /// Accepts `true/false`, `1/0`, `yes/no`, `on/off` — case-insensitive. Anything else
/// returns `default`. /// returns `default`.
pub async fn get_bool(pool: &PgPool, key: &str, default: bool) -> bool { pub async fn get_bool(cache: &ConfigCache, key: &str, default: bool) -> bool {
let Some(raw) = fetch_raw(pool, key).await else { return default }; let Some(raw) = cache.get_raw(key).await else { return default };
match raw.trim().to_ascii_lowercase().as_str() { match raw.trim().to_ascii_lowercase().as_str() {
"true" | "1" | "yes" | "on" => true, "true" | "1" | "yes" | "on" => true,
"false" | "0" | "no" | "off" => false, "false" | "0" | "no" | "off" => false,

145
backend/src/services/disk.rs Normal file
View File

@@ -0,0 +1,145 @@
//! Cached view of the filesystem backing the media directory.
//!
//! Free/total disk space is needed on two hot paths — the per-user storage quota
//! (checked on every upload *and* every `GET /me/quota` poll) and the admin stats
//! endpoint. Reading it means `sysinfo::Disks::new_with_refreshed_list()`, which stats
//! every mounted filesystem; doing that per request is wasteful for a number that
//! barely moves. [`DiskCache`] refreshes it at most once per [`TTL`] and serves the
//! rest from memory.
use std::path::Path;
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
/// How long a disk reading is trusted before the next call re-stats the filesystem.
const TTL: Duration = Duration::from_secs(15);
#[derive(Clone, Copy)]
pub struct DiskInfo {
pub total: u64,
pub free: u64,
}
/// Cheap-to-clone cache of the media filesystem's total/free bytes. Lives in
/// `AppState`.
#[derive(Clone)]
pub struct DiskCache {
inner: Arc<RwLock<Option<(DiskInfo, Instant)>>>,
}
impl DiskCache {
pub fn new() -> Self {
Self {
inner: Arc::new(RwLock::new(None)),
}
}
/// Cached `(total, free)` bytes for the filesystem that holds `media_path`.
///
/// Returns `None` when the mount can't be resolved — callers MUST treat that as
/// "unknown", never "zero free". (The quota path in particular fails *open* on
/// `None`: enforcing a 0-byte limit would lock every user out of uploading.)
pub fn snapshot(&self, media_path: &Path) -> Option<DiskInfo> {
if let Some((info, at)) = *self.inner.read().unwrap() {
if at.elapsed() < TTL {
return Some(info);
}
}
let info = read_disk_for_path(media_path)?;
*self.inner.write().unwrap() = Some((info, Instant::now()));
Some(info)
}
}
impl Default for DiskCache {
fn default() -> Self {
Self::new()
}
}
/// Resolve the filesystem backing `media_path` and read its total/free bytes.
///
/// Snapshots the mount table via `sysinfo`, then delegates the selection to the pure
/// [`select_disk`] so the (fiddly, edge-case-prone) matching logic is unit-testable
/// without touching the real filesystem.
fn read_disk_for_path(media_path: &Path) -> Option<DiskInfo> {
let disks = sysinfo::Disks::new_with_refreshed_list();
let mounts: Vec<(String, u64, u64)> = disks
.iter()
.map(|d| {
(
d.mount_point().to_string_lossy().to_string(),
d.total_space(),
d.available_space(),
)
})
.collect();
select_disk(&mounts, &media_path.to_string_lossy())
}
/// Pick the filesystem for `media_path` from a `(mount_point, total, free)` table.
///
/// Chooses the **longest** mount point that is a prefix of `media_path` (the most
/// specific filesystem) rather than the first match — otherwise the root `/` mount,
/// which prefixes every absolute path, could shadow a dedicated `/media` volume. Falls
/// back to `/` when nothing prefixes the path (e.g. a relative media path), and to
/// `None` when even that is absent — the caller treats `None` as "unknown" and fails
/// open on quota.
fn select_disk(mounts: &[(String, u64, u64)], media_path: &str) -> Option<DiskInfo> {
mounts
.iter()
.filter(|(mp, _, _)| media_path.starts_with(mp.as_str()))
.max_by_key(|(mp, _, _)| mp.len())
.or_else(|| mounts.iter().find(|(mp, _, _)| mp == "/"))
.map(|(_, total, free)| DiskInfo {
total: *total,
free: *free,
})
}
#[cfg(test)]
mod tests {
use super::select_disk;
#[test]
fn picks_longest_matching_mount() {
// Both "/" and "/media" prefix the path; the dedicated volume must win.
let mounts = vec![
("/".to_string(), 100, 40),
("/media".to_string(), 200, 150),
];
let d = select_disk(&mounts, "/media/originals/x.jpg").unwrap();
assert_eq!((d.total, d.free), (200, 150));
}
#[test]
fn falls_back_to_root_when_no_specific_mount_matches() {
let mounts = vec![
("/".to_string(), 100, 40),
("/media".to_string(), 200, 150),
];
// "/var/lib" is only prefixed by "/".
let d = select_disk(&mounts, "/var/lib/data").unwrap();
assert_eq!((d.total, d.free), (100, 40));
}
#[test]
fn relative_path_uses_root_fallback() {
let mounts = vec![("/".to_string(), 100, 40)];
// A relative path prefixes nothing, so the explicit "/" fallback applies.
let d = select_disk(&mounts, "media/originals").unwrap();
assert_eq!((d.total, d.free), (100, 40));
}
#[test]
fn none_when_no_mount_matches_and_no_root() {
// No "/" present and nothing prefixes the relative path → unknown (fail-open).
let mounts = vec![("/data".to_string(), 100, 40)];
assert!(select_disk(&mounts, "relative/path").is_none());
}
#[test]
fn none_on_empty_mount_table() {
assert!(select_disk(&[], "/media/x").is_none());
}
}

96
backend/src/services/export.rs
View File

@@ -192,6 +192,23 @@ async fn run_zip_export(
// ── HTML viewer export ────────────────────────────────────────────────────── // ── HTML viewer export ──────────────────────────────────────────────────────
/// Where a media entry's bytes come from at ZIP-writing time. Derived variants
/// (thumbnails, compressed images) are staged under the temp dir; original-fidelity
/// variants (videos, small images) are streamed straight from the source on disk so
/// the export never transiently doubles disk usage by copying large files to temp.
enum MediaSource {
Temp(PathBuf),
Original(PathBuf),
}
impl MediaSource {
fn path(&self) -> &Path {
match self {
MediaSource::Temp(p) | MediaSource::Original(p) => p,
}
}
}
async fn run_html_export( async fn run_html_export(
event_id: Uuid, event_id: Uuid,
event_name: &str, event_name: &str,
@@ -221,6 +238,9 @@ async fn run_html_export(
// 3. Process media and build post data // 3. Process media and build post data
let mut viewer_posts: Vec<ViewerPost> = Vec::new(); let mut viewer_posts: Vec<ViewerPost> = Vec::new();
// (zip entry name under media/, where its bytes come from). Built here, streamed
// into the ZIP in step 5 — so we also know the exact file count without a rescan.
let mut media_manifest: Vec<(String, MediaSource)> = Vec::new();
for (i, row) in uploads.iter().enumerate() { for (i, row) in uploads.iter().enumerate() {
let src = media_path.join(&row.original_path); let src = media_path.join(&row.original_path);
@@ -231,8 +251,10 @@ async fn run_html_export(
let is_video = row.mime_type.starts_with("video/"); let is_video = row.mime_type.starts_with("video/");
let id_str = row.id.to_string(); let id_str = row.id.to_string();
// Generate thumbnail and full variant // Generate thumbnail and full variant. `full_source` says where the full-res
let (thumb_name, full_name) = if is_video { // bytes come from at ZIP time — for videos and small images that's the original
// on disk (streamed directly, never copied to temp).
let (thumb_name, full_name, full_source) = if is_video {
let thumb = format!("{id_str}_thumb.jpg"); let thumb = format!("{id_str}_thumb.jpg");
let full_ext = ext_from_path(&row.original_path); let full_ext = ext_from_path(&row.original_path);
let full = format!("{id_str}.{full_ext}"); let full = format!("{id_str}.{full_ext}");
@@ -259,14 +281,13 @@ async fn run_html_export(
Ok(output) if output.status.success() => {} Ok(output) if output.status.success() => {}
_ => { _ => {
tracing::warn!("ffmpeg thumbnail failed for upload {}, skipping thumb", row.id); tracing::warn!("ffmpeg thumbnail failed for upload {}, skipping thumb", row.id);
// Create empty thumb entry — viewer handles missing thumbs gracefully // Missing thumb entry — viewer handles missing thumbs gracefully.
} }
} }
// Copy video as-is // Stream the video full-res straight from the original at ZIP time — no
tokio::fs::copy(&src, media_tmp.join(&full)).await?; // copy to temp (that used to transiently double disk usage per video).
(thumb, full, MediaSource::Original(src.clone()))
(thumb, full)
} else { } else {
let thumb = format!("{id_str}_thumb.jpg"); let thumb = format!("{id_str}_thumb.jpg");
let ext = ext_from_path(&row.original_path); let ext = ext_from_path(&row.original_path);
@@ -291,13 +312,12 @@ async fn run_html_export(
tracing::warn!("thumbnail generation failed for upload {}: {e:#}", row.id); tracing::warn!("thumbnail generation failed for upload {}: {e:#}", row.id);
} }
// Full variant: compress if >5MB, otherwise copy original // Full variant: compress to temp if >5MB, otherwise stream the original
// as-is (no temp copy).
let src_meta = tokio::fs::metadata(&src).await?; let src_meta = tokio::fs::metadata(&src).await?;
let full_path = media_tmp.join(&full); let full_source = if src_meta.len() > 5_000_000 {
if src_meta.len() > 5_000_000 {
// Resize to max 2000px
let src_clone = src.clone(); let src_clone = src.clone();
let full_path = media_tmp.join(&full);
let full_path_clone = full_path.clone(); let full_path_clone = full_path.clone();
let compress_result = tokio::task::spawn_blocking(move || -> Result<()> { let compress_result = tokio::task::spawn_blocking(move || -> Result<()> {
@@ -311,17 +331,31 @@ async fn run_html_export(
}) })
.await?; .await?;
if let Err(e) = compress_result { match compress_result {
tracing::warn!("compression failed for upload {}, copying original: {e:#}", row.id); Ok(()) => MediaSource::Temp(full_path),
tokio::fs::copy(&src, &full_path).await?; Err(e) => {
tracing::warn!(
"compression failed for upload {}, using original: {e:#}",
row.id
);
MediaSource::Original(src.clone())
}
} }
} else { } else {
tokio::fs::copy(&src, &full_path).await?; MediaSource::Original(src.clone())
} };
(thumb, full) (thumb, full, full_source)
}; };
// Register this post's two media entries. Thumbnails always come from temp
// (they're freshly generated); the full variant's source was decided above.
media_manifest.push((
thumb_name.clone(),
MediaSource::Temp(media_tmp.join(&thumb_name)),
));
media_manifest.push((full_name.clone(), full_source));
// Build comments for this upload // Build comments for this upload
let post_comments: Vec<ViewerComment> = comments let post_comments: Vec<ViewerComment> = comments
.iter() .iter()
@@ -409,27 +443,23 @@ async fn run_html_export(
update_progress(pool, event_id, "html", 78).await; update_progress(pool, event_id, "html", 78).await;
// Write media files from temp directory // Write media files from the manifest built in step 3. Thumbnails and derived
let mut media_entries = tokio::fs::read_dir(&media_tmp).await?; // image variants stream from temp; video and small-image full variants stream
let mut file_count = 0u32; // straight from the original on disk. Sources that don't exist (e.g. a thumb
// whose ffmpeg step failed) are skipped — the viewer tolerates gaps.
let file_total = media_manifest.len().max(1) as f32;
let mut files_written = 0u32; let mut files_written = 0u32;
// Count files first for (name, source) in &media_manifest {
{ let path = source.path();
let mut counter = tokio::fs::read_dir(&media_tmp).await?; if !path.exists() {
while counter.next_entry().await?.is_some() { continue;
file_count += 1;
} }
}
let file_total = file_count.max(1) as f32;
while let Some(dir_entry) = media_entries.next_entry().await? {
let filename = dir_entry.file_name();
let entry_name = format!("media/{}", filename.to_string_lossy());
let entry_name = format!("media/{name}");
let builder = ZipEntryBuilder::new(entry_name.into(), Compression::Stored); let builder = ZipEntryBuilder::new(entry_name.into(), Compression::Stored);
let mut zip_entry = zip.write_entry_stream(builder).await?; let mut zip_entry = zip.write_entry_stream(builder).await?;
let mut f = tokio::fs::File::open(dir_entry.path()).await?.compat(); let mut f = tokio::fs::File::open(path).await?.compat();
fcopy(&mut f, &mut zip_entry).await?; fcopy(&mut f, &mut zip_entry).await?;
zip_entry.close().await?; zip_entry.close().await?;

1
backend/src/services/mod.rs
View File

@@ -1,5 +1,6 @@
pub mod compression; pub mod compression;
pub mod config; pub mod config;
pub mod disk;
pub mod export; pub mod export;
pub mod jobs; pub mod jobs;
pub mod maintenance; pub mod maintenance;

10
backend/src/state.rs
View File

@@ -3,6 +3,8 @@ use tokio::sync::broadcast;
use crate::config::AppConfig; use crate::config::AppConfig;
use crate::services::compression::CompressionWorker; use crate::services::compression::CompressionWorker;
use crate::services::config::ConfigCache;
use crate::services::disk::DiskCache;
use crate::services::rate_limiter::RateLimiter; use crate::services::rate_limiter::RateLimiter;
use crate::services::sse_tickets::SseTicketStore; use crate::services::sse_tickets::SseTicketStore;
@@ -31,6 +33,11 @@ pub struct AppState {
pub compression: CompressionWorker, pub compression: CompressionWorker,
pub rate_limiter: RateLimiter, pub rate_limiter: RateLimiter,
pub sse_tickets: SseTicketStore, pub sse_tickets: SseTicketStore,
/// In-memory cache in front of the `config` table. Reads go through here; the
/// admin PATCH handler and the test reseed invalidate it after committing.
pub config_cache: ConfigCache,
/// Cached total/free bytes for the media filesystem (quota + admin stats).
pub disk_cache: DiskCache,
} }
impl AppState { impl AppState {
@@ -42,6 +49,7 @@ impl AppState {
config.compression_concurrency, config.compression_concurrency,
sse_tx.clone(), sse_tx.clone(),
); );
let config_cache = ConfigCache::new(pool.clone());
Self { Self {
pool, pool,
config, config,
@@ -49,6 +57,8 @@ impl AppState {
compression, compression,
rate_limiter: RateLimiter::new(), rate_limiter: RateLimiter::new(),
sse_tickets: SseTicketStore::new(), sse_tickets: SseTicketStore::new(),
config_cache,
disk_cache: DiskCache::new(),
} }
} }
} }

64
e2e/specs/06-export/export-video.spec.ts Normal file
View File

@@ -0,0 +1,64 @@
/**
* Covers the HTML export's VIDEO path (perf/stability fix P4). The export used to
* copy every original — including full-size videos — into a temp dir and then re-read
* them into the ZIP (transient 2× disk). It now streams video originals straight from
* disk into the archive. The image-only export specs never exercised that branch, so
* this drives a real video upload → export → and proves the video entry lands in
* Memories.zip (i.e. the streamed-from-original path works and the video isn't dropped).
*
* The fixture clip is <1s, so ffmpeg extracts no thumbnail frame — but exits 0, so the
* compression worker keeps the upload (it isn't auto-cleaned). That's the intended
* shape here: the full video is exported even when its thumbnail is absent.
*/
import { test, expect } from '../../fixtures/test';
import { uploadRaw } from '../../helpers/upload-client';
import { readFileSync } from 'node:fs';
import { join } from 'node:path';
const BASE = process.env.E2E_FRONTEND_URL ?? 'http://localhost:3101';
test.describe('Export — video streaming (P4)', () => {
test('a real video upload is streamed into Memories.zip (not dropped)', async ({ host }) => {
test.setTimeout(60_000);
const bearer = { Authorization: `Bearer ${host.jwt}` };
// Seed a real video upload owned by the host.
const mp4 = readFileSync(join(process.cwd(), 'fixtures', 'media', 'sample.mp4'));
const up = await uploadRaw(host.jwt, mp4, { filename: 'clip.mp4', contentType: 'video/mp4' });
expect(up.status).toBe(201);
const { id } = await up.json();
// Release the gallery → spawns the real zip + html export jobs.
const rel = await fetch(`${BASE}/api/v1/host/gallery/release`, { method: 'POST', headers: bearer });
expect(rel.status).toBe(204);
// Wait for the HTML (Memories.zip) job — that's the one whose media staging P4 changed.
await expect
.poll(
async () => {
const res = await fetch(`${BASE}/api/v1/export/status`, { headers: bearer });
return (await res.json()).html?.status;
},
{ timeout: 45_000, intervals: [500] }
)
.toBe('done');
// Download Memories.zip via the gated single-use ticket.
const ticketRes = await fetch(`${BASE}/api/v1/export/ticket`, { method: 'POST', headers: bearer });
const { ticket } = await ticketRes.json();
const dl = await fetch(`${BASE}/api/v1/export/html?ticket=${encodeURIComponent(ticket)}`);
expect(dl.status).toBe(200);
const bytes = new Uint8Array(await dl.arrayBuffer());
// Valid ZIP (PK local-file-header magic).
expect(Array.from(bytes.subarray(0, 2))).toEqual([0x50, 0x4b]);
// ZIP entry names are stored verbatim (uncompressed) in the archive, so the video's
// media entry name appears as plaintext bytes. Its presence means the streamed
// video entry was written; if the stream had failed, entry.close() would have
// errored and the job would never have reached 'done'.
const needle = `media/${id}.mp4`;
const haystack = new TextDecoder('latin1').decode(bytes);
expect(haystack.includes(needle), `Memories.zip must contain ${needle}`).toBe(true);
});
});

11
frontend/src/lib/sse.ts
View File

@@ -109,6 +109,17 @@ export function connectSse(): void {
); );
} }
// `resync` is emitted by the server when our broadcast subscription fell
// behind and events were dropped. Rather than let those losses leave the feed
// stale, fetch the gap since the last event we actually saw and fan it out as
// a feed-delta (which reconciles new uploads AND deletions). Handled with its
// own listener — not via `dispatch` — so reading `lastEventTime` as the gap
// start isn't clobbered by dispatch bumping it to "now".
eventSource.addEventListener('resync', () => {
const since = lastEventTime;
if (since) void deltaFetchAndFan(since);
});
eventSource.onerror = () => { eventSource.onerror = () => {
// EventSource auto-reconnects but the connection state can stay broken; close // EventSource auto-reconnects but the connection state can stay broken; close
// and try again ourselves with exponential backoff capped at 60s. Prevents // and try again ourselves with exponential backoff capped at 60s. Prevents

5
frontend/src/routes/feed/+page.svelte
View File

@@ -249,6 +249,11 @@
const dead = new Set(delta.deleted_ids); const dead = new Set(delta.deleted_ids);
uploads = uploads.filter((u) => !dead.has(u.id)); uploads = uploads.filter((u) => !dead.has(u.id));
} }
// A delta reconciles new uploads and deletions, but not like/comment
// counts that changed on already-visible cards while we were
// disconnected or lagged (a `resync`). Debounced page-1 refresh merges
// those fresh counts in place without disturbing scroll.
scheduleInPlaceRefresh();
} catch { /* ignore */ } } catch { /* ignore */ }
}) })
); );