feat(cleanup): add quota-aware background cleaner and immediate post-stream file unlinking
This commit is contained in:
parent
29cd9a8688
commit
bbf6e38ba0
5 changed files with 92 additions and 19 deletions
1
backend/Cargo.lock
generated
1
backend/Cargo.lock
generated
|
|
@ -681,6 +681,7 @@ name = "just-download-backend"
|
||||||
version = "2.0.0"
|
version = "2.0.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"axum",
|
"axum",
|
||||||
|
"futures-util",
|
||||||
"mime_guess",
|
"mime_guess",
|
||||||
"regex",
|
"regex",
|
||||||
"reqwest",
|
"reqwest",
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,7 @@ tokio = { version = "1.38", features = ["full"] }
|
||||||
tokio-util = { version = "0.7", features = ["io"] }
|
tokio-util = { version = "0.7", features = ["io"] }
|
||||||
tower = { version = "0.4", features = ["util"] }
|
tower = { version = "0.4", features = ["util"] }
|
||||||
tower-http = { version = "0.5", features = ["cors", "trace", "fs", "compression-full"] }
|
tower-http = { version = "0.5", features = ["cors", "trace", "fs", "compression-full"] }
|
||||||
|
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
|
||||||
serde = { version = "1.0", features = ["derive"] }
|
serde = { version = "1.0", features = ["derive"] }
|
||||||
serde_json = "1.0"
|
serde_json = "1.0"
|
||||||
regex = "1.10"
|
regex = "1.10"
|
||||||
|
|
@ -23,7 +24,7 @@ which = "6.0"
|
||||||
|
|
||||||
[profile.release]
|
[profile.release]
|
||||||
opt-level = 3
|
opt-level = 3
|
||||||
lto = "fat"
|
lto = "thin"
|
||||||
codegen-units = 1
|
codegen-units = 1
|
||||||
panic = "abort"
|
panic = "abort"
|
||||||
strip = true
|
strip = true
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,7 @@ pub struct AppConfig {
|
||||||
pub download_dir: PathBuf,
|
pub download_dir: PathBuf,
|
||||||
pub max_file_age_secs: u64,
|
pub max_file_age_secs: u64,
|
||||||
pub cleanup_interval_secs: u64,
|
pub cleanup_interval_secs: u64,
|
||||||
|
pub max_dir_size_mb: u64,
|
||||||
pub semaphore: Arc<Semaphore>,
|
pub semaphore: Arc<Semaphore>,
|
||||||
pub ytdlp_bin: String,
|
pub ytdlp_bin: String,
|
||||||
}
|
}
|
||||||
|
|
@ -23,20 +24,27 @@ impl AppConfig {
|
||||||
let download_dir = PathBuf::from(
|
let download_dir = PathBuf::from(
|
||||||
std::env::var("DOWNLOAD_DIR").unwrap_or_else(|_| "downloads".to_string()),
|
std::env::var("DOWNLOAD_DIR").unwrap_or_else(|_| "downloads".to_string()),
|
||||||
);
|
);
|
||||||
|
// Automatically delete files older than 60 seconds (stream completes within seconds)
|
||||||
let max_file_age_secs = std::env::var("MAX_FILE_AGE_SECS")
|
let max_file_age_secs = std::env::var("MAX_FILE_AGE_SECS")
|
||||||
.ok()
|
.ok()
|
||||||
.and_then(|v| v.parse().ok())
|
.and_then(|v| v.parse().ok())
|
||||||
.unwrap_or(3600); // 1 hour
|
.unwrap_or(60);
|
||||||
|
// Run background cleaner every 30 seconds
|
||||||
let cleanup_interval_secs = std::env::var("CLEANUP_INTERVAL_SECS")
|
let cleanup_interval_secs = std::env::var("CLEANUP_INTERVAL_SECS")
|
||||||
.ok()
|
.ok()
|
||||||
.and_then(|v| v.parse().ok())
|
.and_then(|v| v.parse().ok())
|
||||||
.unwrap_or(600); // 10 minutes
|
.unwrap_or(30);
|
||||||
|
// Max total directory size: 200 MB quota to prevent disk bloat
|
||||||
|
let max_dir_size_mb = std::env::var("MAX_DIR_SIZE_MB")
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse().ok())
|
||||||
|
.unwrap_or(200);
|
||||||
|
|
||||||
// Limit concurrent downloads on low-memory servers (default 2)
|
// Limit concurrent downloads on low-memory servers (default 2)
|
||||||
let max_concurrent = std::env::var("MAX_CONCURRENT_DOWNLOADS")
|
let max_concurrent = std::env::var("MAX_CONCURRENT_DOWNLOADS")
|
||||||
.ok()
|
.ok()
|
||||||
.and_then(|v| v.parse().ok())
|
.and_then(|v| v.parse().ok())
|
||||||
.unwrap_or(3);
|
.unwrap_or(2);
|
||||||
|
|
||||||
// Find yt-dlp binary
|
// Find yt-dlp binary
|
||||||
let ytdlp_bin = if let Ok(custom) = std::env::var("YTDLP_PATH") {
|
let ytdlp_bin = if let Ok(custom) = std::env::var("YTDLP_PATH") {
|
||||||
|
|
@ -55,6 +63,7 @@ impl AppConfig {
|
||||||
download_dir,
|
download_dir,
|
||||||
max_file_age_secs,
|
max_file_age_secs,
|
||||||
cleanup_interval_secs,
|
cleanup_interval_secs,
|
||||||
|
max_dir_size_mb,
|
||||||
semaphore: Arc::new(Semaphore::new(max_concurrent)),
|
semaphore: Arc::new(Semaphore::new(max_concurrent)),
|
||||||
ytdlp_bin,
|
ytdlp_bin,
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -39,8 +39,9 @@ async fn main() {
|
||||||
let cleanup_dir = config.download_dir.clone();
|
let cleanup_dir = config.download_dir.clone();
|
||||||
let max_age = config.max_file_age_secs;
|
let max_age = config.max_file_age_secs;
|
||||||
let interval = config.cleanup_interval_secs;
|
let interval = config.cleanup_interval_secs;
|
||||||
|
let max_mb = config.max_dir_size_mb;
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
run_cleanup_loop(cleanup_dir, max_age, interval).await;
|
run_cleanup_loop(cleanup_dir, max_age, interval, max_mb).await;
|
||||||
});
|
});
|
||||||
|
|
||||||
// Build Axum routers
|
// Build Axum routers
|
||||||
|
|
@ -123,28 +124,63 @@ async fn health_handler() -> Json<serde_json::Value> {
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Periodic background task that removes files older than max_age_secs
|
/// Periodic background task that removes files older than max_age_secs and enforces directory quota
|
||||||
async fn run_cleanup_loop(dir: PathBuf, max_age_secs: u64, interval_secs: u64) {
|
async fn run_cleanup_loop(dir: PathBuf, max_age_secs: u64, interval_secs: u64, max_dir_size_mb: u64) {
|
||||||
let mut ticker = tokio::time::interval(Duration::from_secs(interval_secs));
|
let mut ticker = tokio::time::interval(Duration::from_secs(interval_secs));
|
||||||
|
let max_bytes = max_dir_size_mb * 1024 * 1024;
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
ticker.tick().await;
|
ticker.tick().await;
|
||||||
|
let mut files: Vec<(PathBuf, u64, std::time::SystemTime)> = Vec::new();
|
||||||
|
let mut total_size: u64 = 0;
|
||||||
|
let now = std::time::SystemTime::now();
|
||||||
|
|
||||||
if let Ok(mut entries) = tokio::fs::read_dir(&dir).await {
|
if let Ok(mut entries) = tokio::fs::read_dir(&dir).await {
|
||||||
let now = std::time::SystemTime::now();
|
|
||||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||||
if let Ok(meta) = entry.metadata().await {
|
if let Ok(meta) = entry.metadata().await {
|
||||||
if meta.is_file() {
|
if meta.is_file() {
|
||||||
if let Ok(modified) = meta.modified() {
|
let path = entry.path();
|
||||||
if let Ok(elapsed) = now.duration_since(modified) {
|
let size = meta.len();
|
||||||
if elapsed.as_secs() > max_age_secs {
|
let modified = meta.modified().unwrap_or(now);
|
||||||
let _ = tokio::fs::remove_file(entry.path()).await;
|
let elapsed = now.duration_since(modified).unwrap_or(Duration::ZERO);
|
||||||
info!("Cleaned up expired file: {:?}", entry.path());
|
|
||||||
}
|
let file_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
|
||||||
}
|
let is_partial = file_name.ends_with(".part")
|
||||||
|
|| file_name.ends_with(".ytdl")
|
||||||
|
|| file_name.ends_with(".tmp")
|
||||||
|
|| file_name.ends_with(".temp");
|
||||||
|
|
||||||
|
// If file is older than max_age or is a stale partial download older than 30s
|
||||||
|
if elapsed.as_secs() > max_age_secs || (is_partial && elapsed.as_secs() > 30) {
|
||||||
|
let _ = tokio::fs::remove_file(&path).await;
|
||||||
|
info!("Auto-cleaner removed expired file: {:?} ({} KB)", path, size / 1024);
|
||||||
|
} else {
|
||||||
|
total_size += size;
|
||||||
|
files.push((path, size, modified));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// If total downloads folder size exceeds quota, delete oldest files until under 50% quota
|
||||||
|
if total_size > max_bytes {
|
||||||
|
info!(
|
||||||
|
"Downloads directory exceeded limit ({} MB / {} MB). Purging oldest files...",
|
||||||
|
total_size / (1024 * 1024),
|
||||||
|
max_dir_size_mb
|
||||||
|
);
|
||||||
|
files.sort_by_key(|f| f.2);
|
||||||
|
let target_size = max_bytes / 2;
|
||||||
|
for (path, size, _) in files {
|
||||||
|
if total_size <= target_size {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
let _ = tokio::fs::remove_file(&path).await;
|
||||||
|
info!("Purged file to free space: {:?}", path);
|
||||||
|
total_size = total_size.saturating_sub(size);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -6,19 +6,45 @@ use axum::body::Body;
|
||||||
use axum::http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
|
use axum::http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
|
||||||
use axum::http::HeaderMap;
|
use axum::http::HeaderMap;
|
||||||
use axum::response::{IntoResponse, Response};
|
use axum::response::{IntoResponse, Response};
|
||||||
|
use futures_util::StreamExt;
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
|
use std::sync::Arc;
|
||||||
use tokio_util::io::ReaderStream;
|
use tokio_util::io::ReaderStream;
|
||||||
|
use tracing::info;
|
||||||
|
|
||||||
use crate::error::AppError;
|
use crate::error::AppError;
|
||||||
|
|
||||||
/// Streams a file response with Content-Disposition and schedules background cleanup
|
struct FileCleaner {
|
||||||
|
path: PathBuf,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for FileCleaner {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
let path = self.path.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let _ = tokio::fs::remove_file(&path).await;
|
||||||
|
info!("Auto-cleaner unlinked served file immediately: {:?}", path);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Streams a file response with Content-Disposition and guarantees immediate cleanup
|
||||||
pub async fn stream_file_response(path: PathBuf, filename: String) -> Result<Response, AppError> {
|
pub async fn stream_file_response(path: PathBuf, filename: String) -> Result<Response, AppError> {
|
||||||
let file = tokio::fs::File::open(&path).await.map_err(|e| {
|
let file = tokio::fs::File::open(&path).await.map_err(|e| {
|
||||||
AppError::NotFound(format!("Failed to open generated file: {}", e))
|
AppError::NotFound(format!("Failed to open generated file: {}", e))
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
let file_len = file.metadata().await.ok().map(|m| m.len());
|
let file_len = file.metadata().await.ok().map(|m| m.len());
|
||||||
let stream = ReaderStream::new(file);
|
let cleaner = Arc::new(FileCleaner {
|
||||||
|
path: path.clone(),
|
||||||
|
});
|
||||||
|
|
||||||
|
let reader_stream = ReaderStream::new(file);
|
||||||
|
// Bind the cleaner Arc to the stream so the file is unlinked immediately when stream ends or client disconnects
|
||||||
|
let stream = reader_stream.map(move |chunk| {
|
||||||
|
let _g = &cleaner;
|
||||||
|
chunk
|
||||||
|
});
|
||||||
let body = Body::from_stream(stream);
|
let body = Body::from_stream(stream);
|
||||||
|
|
||||||
let mime = mime_guess::from_path(&filename)
|
let mime = mime_guess::from_path(&filename)
|
||||||
|
|
@ -36,10 +62,10 @@ pub async fn stream_file_response(path: PathBuf, filename: String) -> Result<Res
|
||||||
headers.insert(CONTENT_DISPOSITION, cd);
|
headers.insert(CONTENT_DISPOSITION, cd);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Background deletion after 15 minutes (900 seconds)
|
// Safety fallback: ensure deletion after 60 seconds even if stream somehow hangs
|
||||||
let path_clone = path.clone();
|
let path_clone = path.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tokio::time::sleep(tokio::time::Duration::from_secs(900)).await;
|
tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
|
||||||
let _ = tokio::fs::remove_file(&path_clone).await;
|
let _ = tokio::fs::remove_file(&path_clone).await;
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue