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

perf(backend): close config/upload/export hot paths + stability fixes

Browse Source 
Performance:
- Cache the runtime `config` table in-memory (ConfigCache) with synchronous
  invalidation on every write (admin PATCH + test reseed). Was re-reading each
  key from Postgres on every request (~8 round-trips per upload).
- Stream uploads chunk-by-chunk to a temp file instead of buffering the whole
  body in RAM (peak was up to the per-class cap, e.g. 500 MB/video); only 512
  sniff-bytes are kept for magic-byte detection, then atomic rename into place.
- Cache the media-filesystem disk snapshot (DiskCache, 15s TTL) shared by the
  quota check and admin stats; drop the discarded System::refresh_all().
- HTML export streams video (and small-image) originals straight into the ZIP
  via a manifest instead of copying them to a temp dir first (removed the
  transient 2x disk usage) and drops the double directory scan.
- Auth extractor resolves session -> live user in one JOIN (was two queries),
  touching last_seen_at by token hash.

Stability:
- SSE: on broadcast lag, emit a `resync` event so the client runs a delta
  fetch instead of silently losing events; frontend reconciles adds, deletions,
  and (via an in-place refresh) like/comment counts on visible cards.
- Storage quota fails OPEN when the disk can't be read (was a 0-byte limit that
  locked out all uploads).
- Graceful shutdown drains in-flight requests on SIGTERM/SIGINT, bounded by a
  10s backstop so open SSE streams can't stall a deploy.
- Upload removes the persisted file if the DB transaction fails (no orphaned
  bytes with no row to reclaim them).

Tests:
- New pure select_disk() with 5 unit tests (longest-prefix, fallbacks, fail-open).
- New e2e export-video spec covering the HTML export's video-streaming branch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
fabi
2026-07-07 20:26:45 +02:00
parent 4cdb3ae14a
commit d6c91974eb
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>,
) -> Result<(StatusCode, Json<JoinResponse>), AppError> {
let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
let join_rate_on = config::get_bool(&state.pool, "join_rate_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.config_cache, "join_rate_enabled", true).await;
if rate_limits_on && join_rate_on
&& !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)
// softens that into a real cost.
let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
let recover_rate_on = config::get_bool(&state.pool, "recover_rate_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.config_cache, "recover_rate_enabled", true).await;
if rate_limits_on && recover_rate_on {
let name_key = display_name.to_lowercase();
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
// honest typos.
let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
let admin_rate_on = config::get_bool(&state.pool, "admin_login_rate_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.config_cache, "admin_login_rate_enabled", true).await;
if rate_limits_on && admin_rate_on
&& !state.rate_limiter.check(
format!("admin_login:{ip}"),

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

@@ -37,34 +37,32 @@ impl FromRequestParts<AppState> for AuthUser {
.strip_prefix("Bearer ")
.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()))?;
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
.map_err(|e| AppError::Internal(e.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
// non-fatal but worth surfacing — silent swallowing hides DB connection
// pressure that would otherwise be the first symptom of a real problem.
let pool = state.pool.clone();
let session_id = session.id;
let touch_hash = token_hash.clone();
tokio::spawn(async move {
if let Err(e) = Session::touch(&pool, session_id).await {
tracing::warn!(error = ?e, session_id = %session_id, "session touch failed");
if let Err(e) = Session::touch_by_token_hash(&pool, &touch_hash).await {
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::Json;
use serde::{Deserialize, Serialize};
use sysinfo::System;
use crate::auth::middleware::RequireAdmin;
use crate::error::AppError;
@@ -68,23 +67,12 @@ pub async fn get_stats(
.fetch_one(&state.pool)
.await?;
// Disk usage via sysinfo
let mut sys = System::new();
sys.refresh_all();
let media_path = state.config.media_path.to_string_lossy().to_string();
let (disk_total, disk_free) = sysinfo::Disks::new_with_refreshed_list()
.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))
});
// Disk usage from the shared cache (unknown mount → zeros, same as before).
let (disk_total, disk_free) = state
.disk_cache
.snapshot(&state.config.media_path)
.map(|d| (d.total, d.free))
.unwrap_or((0, 0));
let disk_used = disk_total.saturating_sub(disk_free);
@@ -216,6 +204,11 @@ pub async fn patch_config(
}
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
// (e.g. the privacy note in My Account) refresh without a manual reload.
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
/// each request.
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 export_rate_on = config::get_bool(&state.pool, "export_rate_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.config_cache, "export_rate_enabled", true).await;
if !(rate_limits_on && export_rate_on) {
return Ok(());
}
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
.rate_limiter
.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>,
) -> Result<Json<FeedResponse>, AppError> {
let ip = client_ip(&headers, "unknown");
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
let feed_rate_on = config::get_bool(&state.pool, "feed_rate_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.config_cache, "feed_rate_enabled", true).await;
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
.rate_limiter
.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?
.ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;
let privacy_note = config::get_str(&state.pool, "privacy_note", "").await;
let quota_enabled = config::get_bool(&state.pool, "quota_enabled", true).await;
let storage_quota_enabled = config::get_bool(&state.pool, "storage_quota_enabled", true).await;
let privacy_note = config::get_str(&state.config_cache, "privacy_note", "").await;
let quota_enabled = config::get_bool(&state.config_cache, "quota_enabled", true).await;
let storage_quota_enabled = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await;
Ok(Json(MeContextDto {
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 futures::stream::Stream;
use serde::{Deserialize, Serialize};
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::StreamExt;
@@ -62,7 +63,14 @@ pub async fn stream(
Ok(sse_event) => Some(Ok(Event::default()
.event(sse_event.event_type)
.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(

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.
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)
}

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

@@ -43,10 +43,10 @@ pub async fn upload(
mut multipart: Multipart,
) -> Result<(StatusCode, Json<UploadDto>), AppError> {
// 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 upload_rate_on = config::get_bool(&state.pool, "upload_rate_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.config_cache, "upload_rate_enabled", true).await;
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(
format!("upload:{}", auth.user_id),
upload_rate,
@@ -79,57 +79,85 @@ pub async fn upload(
}
// Read config limits from DB
let max_image_mb: i64 = config::get_i64(&state.pool, "max_image_size_mb", 20).await;
let max_video_mb: i64 = config::get_i64(&state.pool, "max_video_size_mb", 500).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.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 hashtags_csv: Option<String> = None;
while let Some(field) = multipart.next_field().await.map_err(|e| AppError::BadRequest(e.to_string()))? {
let name = field.name().unwrap_or_default().to_string();
match name.as_str() {
"file" => {
// Note: the client-declared filename and Content-Type do NOT determine
// the stored MIME/extension — those come from the file's magic bytes
// below. The declared type is used only to pick a memory cap so an
// oversized body can't be fully buffered before the size check. A
// mislabelled type only makes the cap *stricter* (safe); the
// authoritative per-class check still runs on the detected type.
let declared = field.content_type().unwrap_or("").to_string();
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
};
file_data = Some(read_field_capped(field, cap_bytes).await?);
// Wrap the multipart read so any error after the temp file is created still cleans
// it up (a mid-stream parse failure must not leave a stray `.tmp` on disk).
let parse_result: Result<(), AppError> = async {
while let Some(field) = multipart
.next_field()
.await
.map_err(|e| AppError::BadRequest(e.to_string()))?
{
let name = field.name().unwrap_or_default().to_string();
match name.as_str() {
"file" => {
// The client-declared Content-Type does NOT determine the stored
// MIME/extension — those come from the file's magic bytes below. The
// declared type only picks the streaming cap so an oversized body is
// aborted early; a mislabelled type only makes the cap *stricter*
// (safe), and the authoritative per-class check still runs on the
// detected type.
let declared = field.content_type().unwrap_or("").to_string();
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()))?;
let size = data.len() as i64;
// From here on the temp file may exist; every validation failure removes it before
// 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
// "Zeichen" wording in the error message — `.len()` would be bytes and
// reject perfectly valid German/emoji captions early.
if let Some(ref cap) = caption {
if cap.chars().count() > MAX_CAPTION_LENGTH {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::BadRequest(format!(
"Beschreibung ist zu lang. Maximum: {} Zeichen.",
MAX_CAPTION_LENGTH
@@ -142,26 +170,38 @@ pub async fn upload(
// 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
// client-supplied values.
let kind = infer::get(&data)
.ok_or_else(|| AppError::BadRequest("Dateityp nicht erkannt oder nicht unterstützt.".into()))?;
let (mime, ext) = ALLOWED_MEDIA
let kind = match infer::get(&head) {
Some(k) => k,
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()
.find(|(allowed, _)| *allowed == kind.mime_type())
.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: {}.",
kind.mime_type()
))
})?;
)));
}
};
// Validate file size
// Validate file size against the authoritative per-detected-class limit.
let max_bytes = if mime.starts_with("video/") {
max_video_mb * 1024 * 1024
} else {
max_image_mb * 1024 * 1024
};
if size > max_bytes {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::BadRequest(format!(
"Datei ist zu groß. Maximum: {} MB.",
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
// number of active uploaders. Gated by master + per-area toggles so the admin can
// disable it on trusted instances.
let quota_on = config::get_bool(&state.pool, "quota_enabled", true).await;
let storage_quota_on = config::get_bool(&state.pool, "storage_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.config_cache, "storage_quota_enabled", true).await;
if quota_on && storage_quota_on {
let estimate = compute_storage_quota(&state).await;
if let Some(limit) = estimate.limit_bytes {
let prospective_total = user.total_upload_bytes.saturating_add(size);
if prospective_total > limit {
let _ = tokio::fs::remove_file(&temp_abs).await;
return Err(AppError::TooManyRequests(
"Du hast dein Upload-Limit für dieses Event erreicht.".into(),
None,
@@ -186,16 +227,13 @@ pub async fn upload(
}
}
let upload_id = Uuid::new_v4();
let event_slug = &state.config.event_slug;
// All checks passed — atomically move the temp file to its final, extension-correct
// path (same directory, so the rename is cheap and atomic).
let relative_path = format!("originals/{event_slug}/{upload_id}.{ext}");
let absolute_path = state.config.media_path.join(&relative_path);
// Ensure directory exists and write file
if let Some(parent) = absolute_path.parent() {
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()))?;
tokio::fs::rename(&temp_abs, &absolute_path)
.await
.map_err(|e| AppError::Internal(e.into()))?;
// Process hashtags from caption and explicit CSV
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
// crash between the bytes increment and the insert would permanently charge
// bytes with no row to reclaim them (silent quota erosion / spurious lockout).
let mut tx = state.pool.begin().await?;
sqlx::query("UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1")
.bind(auth.user_id)
.bind(size)
.execute(&mut *tx)
let tx_result: Result<Upload, AppError> = async {
let mut tx = state.pool.begin().await?;
sqlx::query("UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1")
.bind(auth.user_id)
.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?;
let upload = Upload::create(
&mut *tx,
auth.event_id,
auth.user_id,
&relative_path,
&mime,
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?;
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?;
Ok(upload)
}
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
state
@@ -339,30 +391,68 @@ pub async fn delete_upload(
Ok(StatusCode::NO_CONTENT)
}
/// Read a multipart field into memory, aborting with a 400 the moment it exceeds
/// `max_bytes`. Without this the whole field is buffered (up to the HTTP body cap)
/// before the post-read size check runs, so a request claiming to be a tiny image
/// could still force hundreds of MB of allocation. Streaming with an early abort
/// bounds peak memory to roughly the applicable per-class limit.
async fn read_field_capped(
/// Number of leading bytes retained in memory for magic-byte (`infer`) sniffing. Every
/// allowed type's signature sits well within this; 512 is comfortably generous.
const HEAD_SNIFF_BYTES: usize = 512;
/// Stream a multipart field straight to `dest`, aborting with a 400 the moment it
/// 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<'_>,
dest: &std::path::Path,
max_bytes: usize,
) -> Result<Vec<u8>, AppError> {
let mut buf: Vec<u8> = Vec::new();
while let Some(chunk) = field
.chunk()
) -> Result<(i64, Vec<u8>), AppError> {
use tokio::io::AsyncWriteExt;
let mut file = tokio::fs::File::create(dest)
.await
.map_err(|e| AppError::BadRequest(format!("Datei konnte nicht gelesen werden: {e}")))?
{
if buf.len().saturating_add(chunk.len()) > max_bytes {
.map_err(|e| AppError::Internal(e.into()))?;
let mut total: usize = 0;
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!(
"Datei ist zu groß. Maximum: {} MB.",
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.
@@ -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
/// check (upload handler) or hide the UI (quota endpoint).
pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate {
let quota_on = config::get_bool(&state.pool, "quota_enabled", true).await;
let storage_quota_on = config::get_bool(&state.pool, "storage_quota_enabled", true).await;
let tolerance = config::get_f64(&state.pool, "quota_tolerance", 0.75).await;
let quota_on = config::get_bool(&state.config_cache, "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.config_cache, "quota_tolerance", 0.75).await;
let (active_count,): (i64,) = sqlx::query_as(
"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,));
let active = active_count.max(1);
let media_path = state.config.media_path.to_string_lossy().to_string();
let free_disk = sysinfo::Disks::new_with_refreshed_list()
.iter()
.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;
// Cached disk reading. `None` means we couldn't resolve the media filesystem.
let disk = state.disk_cache.snapshot(&state.config.media_path);
let free_disk = disk.map(|d| d.free as i64).unwrap_or(0);
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 {
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?;
tracing::info!("listening on {}", listener.local_addr()?);
axum::serve(listener, router).await?;
axum::serve(listener, router)
.with_graceful_shutdown(shutdown_signal())
.await?;
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
}
pub async fn touch(pool: &PgPool, id: Uuid) -> Result<(), sqlx::Error> {
sqlx::query("UPDATE session SET last_seen_at = NOW() WHERE id = $1")
.bind(id)
/// Resolve a session token straight to its live user row in one round-trip.
///
/// 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)
.await?;
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
//! 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
//! 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
//! (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;
async fn fetch_raw(pool: &PgPool, key: &str) -> Option<String> {
sqlx::query_as::<_, (String,)>("SELECT value FROM config WHERE key = $1")
.bind(key)
.fetch_optional(pool)
.await
.ok()
.flatten()
.map(|(v,)| v)
/// How long a loaded snapshot is trusted before the next read reloads it. This is a
/// backstop for out-of-band DB changes only — every in-process write invalidates the
/// cache synchronously, so tests that PATCH-then-assert never depend on this expiring.
const RELOAD_TTL: Duration = Duration::from_secs(30);
struct Snapshot {
values: HashMap<String, String>,
loaded_at: Instant,
}
pub async fn get_str(pool: &PgPool, key: &str, default: &str) -> String {
fetch_raw(pool, key).await.unwrap_or_else(|| default.to_string())
/// In-memory cache of the entire `config` table. Cheap to `clone` (shares the pool and
/// 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 {
fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default)
impl ConfigCache {
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 {
fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default)
pub async fn get_str(cache: &ConfigCache, key: &str, default: &str) -> String {
cache.get_raw(key).await.unwrap_or_else(|| default.to_string())
}
pub async fn get_f64(pool: &PgPool, key: &str, default: f64) -> f64 {
fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default)
pub async fn get_i64(cache: &ConfigCache, key: &str, default: i64) -> i64 {
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.
/// Accepts `true/false`, `1/0`, `yes/no`, `on/off` — case-insensitive. Anything else
/// returns `default`.
pub async fn get_bool(pool: &PgPool, key: &str, default: bool) -> bool {
let Some(raw) = fetch_raw(pool, key).await else { return default };
pub async fn get_bool(cache: &ConfigCache, key: &str, default: bool) -> bool {
let Some(raw) = cache.get_raw(key).await else { return default };
match raw.trim().to_ascii_lowercase().as_str() {
"true" | "1" | "yes" | "on" => true,
"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 ──────────────────────────────────────────────────────
/// 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(
event_id: Uuid,
event_name: &str,
@@ -221,6 +238,9 @@ async fn run_html_export(
// 3. Process media and build post data
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() {
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 id_str = row.id.to_string();
// Generate thumbnail and full variant
let (thumb_name, full_name) = if is_video {
// Generate thumbnail and full variant. `full_source` says where the full-res
// 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 full_ext = ext_from_path(&row.original_path);
let full = format!("{id_str}.{full_ext}");
@@ -259,14 +281,13 @@ async fn run_html_export(
Ok(output) if output.status.success() => {}
_ => {
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
tokio::fs::copy(&src, media_tmp.join(&full)).await?;
(thumb, full)
// Stream the video full-res straight from the original at ZIP time — no
// copy to temp (that used to transiently double disk usage per video).
(thumb, full, MediaSource::Original(src.clone()))
} else {
let thumb = format!("{id_str}_thumb.jpg");
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);
}
// 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 full_path = media_tmp.join(&full);
if src_meta.len() > 5_000_000 {
// Resize to max 2000px
let full_source = if src_meta.len() > 5_000_000 {
let src_clone = src.clone();
let full_path = media_tmp.join(&full);
let full_path_clone = full_path.clone();
let compress_result = tokio::task::spawn_blocking(move || -> Result<()> {
@@ -311,17 +331,31 @@ async fn run_html_export(
})
.await?;
if let Err(e) = compress_result {
tracing::warn!("compression failed for upload {}, copying original: {e:#}", row.id);
tokio::fs::copy(&src, &full_path).await?;
match compress_result {
Ok(()) => MediaSource::Temp(full_path),
Err(e) => {
tracing::warn!(
"compression failed for upload {}, using original: {e:#}",
row.id
);
MediaSource::Original(src.clone())
}
}
} 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
let post_comments: Vec<ViewerComment> = comments
.iter()
@@ -409,27 +443,23 @@ async fn run_html_export(
update_progress(pool, event_id, "html", 78).await;
// Write media files from temp directory
let mut media_entries = tokio::fs::read_dir(&media_tmp).await?;
let mut file_count = 0u32;
// Write media files from the manifest built in step 3. Thumbnails and derived
// image variants stream from temp; video and small-image full variants stream
// 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;
// Count files first
{
let mut counter = tokio::fs::read_dir(&media_tmp).await?;
while counter.next_entry().await?.is_some() {
file_count += 1;
for (name, source) in &media_manifest {
let path = source.path();
if !path.exists() {
continue;
}
}
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 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?;
zip_entry.close().await?;

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

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

10
backend/src/state.rs
View File

@@ -3,6 +3,8 @@ use tokio::sync::broadcast;
use crate::config::AppConfig;
use crate::services::compression::CompressionWorker;
use crate::services::config::ConfigCache;
use crate::services::disk::DiskCache;
use crate::services::rate_limiter::RateLimiter;
use crate::services::sse_tickets::SseTicketStore;
@@ -31,6 +33,11 @@ pub struct AppState {
pub compression: CompressionWorker,
pub rate_limiter: RateLimiter,
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 {
@@ -42,6 +49,7 @@ impl AppState {
config.compression_concurrency,
sse_tx.clone(),
);
let config_cache = ConfigCache::new(pool.clone());
Self {
pool,
config,
@@ -49,6 +57,8 @@ impl AppState {
compression,
rate_limiter: RateLimiter::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 auto-reconnects but the connection state can stay broken; close
// 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);
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 */ }
})
);