diff --git a/.dockerignore b/.dockerignore index ee429ec2..e93c91f9 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,6 +1,10 @@ .git .gitignore downloads +venv +.venv +tests +.superpowers __pycache__ *.pyc *.pyo diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..dfdb8b77 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +*.sh text eol=lf diff --git a/.gitignore b/.gitignore index 148142d8..fbbf8404 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ venv/ +.venv/ downloads/ __pycache__/ *.pyc @@ -7,4 +8,4 @@ assets/*.mov .env .env.local .env.production -.env.development \ No newline at end of file +.env.development diff --git a/README.md b/README.md index f67212c7..b6e0c3c0 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,9 @@ https://github.com/user-attachments/assets/419d3e50-c933-444b-8cab-a9724986ba05 - Bulk downloads — paste multiple URLs at once - Automatic URL deduplication - Clean, responsive UI — no frameworks, no build step -- Single Python file backend (~150 lines) +- Live download progress, speed, size and ETA +- Resumable task records for interrupted downloads +- Cancel active downloads while preserving their partial files for later continuation ## Quick Start @@ -28,14 +30,17 @@ cd reclip ./reclip.sh ``` -Open **http://localhost:8899**. +Open **http://127.0.0.1:8899**. Or with Docker: ```bash -docker build -t reclip . && docker run -p 8899:8899 reclip +docker compose up --build -d ``` +The Compose configuration publishes the service on `127.0.0.1:8899`, so it is +available only on the local computer by default. + ## Usage 1. Paste one or more video URLs into the input box @@ -44,6 +49,36 @@ docker build -t reclip . && docker run -p 8899:8899 reclip 4. Select quality/resolution if available 5. Click **Download** on individual videos, or **Download All** +### Interrupted downloads + +Downloads are stored under the Docker volume mounted at `/app/downloads`. +Task records and resumable media are kept in `.reclip/jobs.sqlite3` and +`.reclip/jobs//`. If a download fails or the container is restarted, +open the page again and use **Continue** on the interrupted task. ReClip checks +the saved media identity before asking yt-dlp to reuse its partial files. + +Continuation depends on the source and media protocol. If the source changes, +does not support byte ranges or no longer exposes the selected format, ReClip +will refuse to append to the old task; use **Restart** to create a new task. +The application does not automatically start interrupted downloads after a +container restart. Completed files can be saved again from the task list. + +### Cancelling downloads + +Click **Cancel** on an active task to stop its current downloader process. The +task first shows **Cancelling** while ReClip confirms that the process and its +children have stopped; only then is it marked **Cancelled**. Partial files and +the last known progress are preserved, and no completed file is published. + +For a cancelled task, **Continue** creates the next attempt in the same task +and asks yt-dlp to reuse compatible partial files. **Restart** creates a new +task from the original URL. **Delete** removes the task record and its task +directory after confirmation. + +Task history now retains the source URL, title and partial files until the task +is explicitly deleted. Do not expose this local service to the public network; +the first version has no user authentication. + ## Supported Sites Anything [yt-dlp supports](https://github.com/yt-dlp/yt-dlp/blob/master/supportedsites.md), including: @@ -52,10 +87,10 @@ YouTube, TikTok, Instagram, Twitter/X, Reddit, Facebook, Vimeo, Twitch, Dailymot ## Stack -- **Backend:** Python + Flask (~150 lines) +- **Backend:** Python + Flask with a SQLite task store - **Frontend:** Vanilla HTML/CSS/JS (single file, no build step) - **Download engine:** [yt-dlp](https://github.com/yt-dlp/yt-dlp) + [ffmpeg](https://ffmpeg.org/) -- **Dependencies:** 2 (Flask, yt-dlp) +- **Dependencies:** Flask, yt-dlp, ffmpeg and the Python standard library ## Disclaimer diff --git a/app.py b/app.py index 2b5eaa5b..612ecdc1 100644 --- a/app.py +++ b/app.py @@ -1,16 +1,140 @@ import os -import uuid import glob import json import subprocess import threading +import time +from urllib.parse import urlparse from flask import Flask, request, jsonify, send_file, render_template +from download_process import run_streaming_process +from job_service import CancelUnavailableError, JobService +from job_store import JobStore +from progress import normalize_download_progress, parse_progress_line +from runtime_guard import RuntimeGuard + app = Flask(__name__) DOWNLOAD_DIR = os.path.join(os.path.dirname(__file__), "downloads") os.makedirs(DOWNLOAD_DIR, exist_ok=True) jobs = {} +jobs_lock = threading.RLock() +DOWNLOAD_TIMEOUT = 300 +_service_lock = threading.RLock() +_job_service = None +_runtime_guard = None + +DOWNLOAD_PROGRESS_TEMPLATE = ( + "download:RECLIP_PROGRESS " + "%(progress.{status,downloaded_bytes,total_bytes,total_bytes_estimate,speed,eta})j" +) +POSTPROCESS_PROGRESS_TEMPLATE = ( + "postprocess:RECLIP_POSTPROCESS " + "%(progress.{status,postprocessor})j" +) + + +def _get_job_service(): + """Initialize the durable service once for the container process.""" + global _job_service, _runtime_guard + if _job_service is not None: + return _job_service + with _service_lock: + if _job_service is not None: + return _job_service + root = os.environ.get("RECLIP_DOWNLOAD_DIR", DOWNLOAD_DIR) + root_path = os.path.abspath(root) + internal = os.path.join(root_path, ".reclip") + guard = RuntimeGuard(os.path.join(internal, "runtime.lock")) + epoch = guard.acquire() + try: + store = JobStore(os.path.join(internal, "jobs.sqlite3")) + store.initialize() + service = JobService(store, root_path, epoch) + service.recover() + except Exception: + guard.close() + raise + _runtime_guard = guard + _job_service = service + return service + + +def _service_error(exc): + if isinstance(exc, KeyError): + return jsonify({"error": "Job not found"}), 404 + if isinstance(exc, ValueError): + return jsonify({"error": str(exc)}), 400 + if isinstance(exc, FileNotFoundError): + return jsonify({"error": "File is no longer available"}), 410 + if isinstance(exc, RuntimeError): + message = str(exc) + status = ( + 503 + if isinstance(exc, CancelUnavailableError) + or "restart" in message.lower() + or "storage" in message.lower() + else 409 + ) + return jsonify({"error": message}), status + return jsonify({"error": "Request could not be completed"}), 500 + + +def _empty_progress(): + return normalize_download_progress({}, now=None) + + +def apply_progress_event(job_id, event, now=None): + """Apply one parsed event without changing a terminal job.""" + if now is None: + now = time.time() + + with jobs_lock: + job = jobs.get(job_id) + if not job or job["status"] in ("done", "error"): + return + + data = event.get("data", {}) if isinstance(event, dict) else {} + if not isinstance(data, dict): + return + + if event.get("kind") == "download": + if data.get("status") == "downloading": + job["phase"] = "downloading" + job["progress"] = normalize_download_progress(data, now) + elif data.get("status") == "finished": + progress = normalize_download_progress(data, now) + progress["speed_bps"] = None + progress["eta_seconds"] = None + job["phase"] = "finalizing" + job["progress"] = progress + elif event.get("kind") == "postprocess": + job["phase"] = "processing" + job["progress"] = None + + +def _mark_job_error(job_id, message): + with jobs_lock: + job = jobs.get(job_id) + if job and job["status"] not in ("done", "error"): + job["status"] = "error" + job["phase"] = "failed" + job["error"] = message + job["progress"] = None + + +def is_safe_url(url): + """Reject anything that isn't a plain http(s) URL. + + This also blocks strings starting with ``-``/``--`` which yt-dlp would + otherwise parse as CLI options (e.g. ``--exec``), letting a caller + smuggle arbitrary flags into the subprocess invocation. + """ + try: + parsed = urlparse(url) + except ValueError: + return False + return parsed.scheme in ("http", "https") and bool(parsed.netloc) def parse_ytdlp_json(stdout): @@ -30,10 +154,22 @@ def parse_ytdlp_json(stdout): def run_download(job_id, url, format_choice, format_id): - job = jobs[job_id] out_template = os.path.join(DOWNLOAD_DIR, f"{job_id}.%(ext)s") - cmd = ["yt-dlp", "--no-playlist", "-o", out_template] + cmd = [ + "yt-dlp", + "--no-playlist", + "--newline", + "--progress", + "--progress-delta", + "0.5", + "--progress-template", + DOWNLOAD_PROGRESS_TEMPLATE, + "--progress-template", + POSTPROCESS_PROGRESS_TEMPLATE, + "-o", + out_template, + ] if format_choice == "audio": cmd += ["-x", "--audio-format", "mp3"] @@ -42,19 +178,42 @@ def run_download(job_id, url, format_choice, format_id): else: cmd += ["-f", "bestvideo+bestaudio/best", "--merge-output-format", "mp4"] - cmd.append(url) + # "--" stops yt-dlp from treating a URL that begins with "-" as an + # option (e.g. "--exec=..."), which would otherwise allow arbitrary + # command execution. + cmd += ["--", url] + + last_error_lines = [] + + def handle_line(line): + event = parse_progress_line(line) + if event is not None: + apply_progress_event(job_id, event) + return + + if line.startswith("ERROR:") or line.startswith("WARNING:"): + last_error_lines.append(line[:1000]) + del last_error_lines[:-20] try: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) - if result.returncode != 0: - job["status"] = "error" - job["error"] = result.stderr.strip().split("\n")[-1] + with jobs_lock: + if job_id not in jobs: + return + jobs[job_id]["phase"] = "preparing" + + returncode = run_streaming_process( + cmd, + handle_line, + timeout_seconds=DOWNLOAD_TIMEOUT, + ) + if returncode != 0: + message = last_error_lines[-1] if last_error_lines else f"yt-dlp exited with code {returncode}" + _mark_job_error(job_id, message.replace("ERROR: ", "")) return files = glob.glob(os.path.join(DOWNLOAD_DIR, f"{job_id}.*")) if not files: - job["status"] = "error" - job["error"] = "Download completed but no file was found" + _mark_job_error(job_id, "Download completed but no file was found") return if format_choice == "audio": @@ -71,22 +230,43 @@ def run_download(job_id, url, format_choice, format_id): except OSError: pass - job["status"] = "done" - job["file"] = chosen ext = os.path.splitext(chosen)[1] - title = job.get("title", "").strip() # Sanitize title for filename + with jobs_lock: + job = jobs.get(job_id) + if not job or job["status"] == "error": + return + title = job.get("title", "").strip() if title: safe_title = "".join(c for c in title if c not in r'\/:*?"<>|').strip()[:100].strip() - job["filename"] = f"{safe_title}{ext}" if safe_title else os.path.basename(chosen) + filename = f"{safe_title}{ext}" if safe_title else os.path.basename(chosen) else: - job["filename"] = os.path.basename(chosen) + filename = os.path.basename(chosen) + with jobs_lock: + job = jobs.get(job_id) + if not job or job["status"] == "error": + return + complete_progress = _empty_progress() + complete_progress["percent"] = 100.0 + job.update( + { + "status": "done", + "phase": "complete", + "progress": complete_progress, + "file": chosen, + "filename": filename, + } + ) except subprocess.TimeoutExpired: - job["status"] = "error" - job["error"] = "Download timed out (5 min limit)" + _mark_job_error(job_id, "Download timed out (5 min limit)") + # The process runner has already terminated the child process tree. + for f in glob.glob(os.path.join(DOWNLOAD_DIR, f"{job_id}.*")): + try: + os.remove(f) + except OSError: + pass except Exception as e: - job["status"] = "error" - job["error"] = str(e) + _mark_job_error(job_id, str(e)) @app.route("/") @@ -100,8 +280,10 @@ def get_info(): url = data.get("url", "").strip() if not url: return jsonify({"error": "No URL provided"}), 400 + if not is_safe_url(url): + return jsonify({"error": "Invalid URL"}), 400 - cmd = ["yt-dlp", "--no-playlist", "-j", url] + cmd = ["yt-dlp", "--no-playlist", "-j", "--", url] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: @@ -146,8 +328,10 @@ def get_playlist_info(): url = data.get("url", "").strip() if not url: return jsonify({"error": "No URL provided"}), 400 + if not is_safe_url(url): + return jsonify({"error": "Invalid URL"}), 400 - cmd = ["yt-dlp", "--flat-playlist", "-J", url] + cmd = ["yt-dlp", "--flat-playlist", "-J", "--", url] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: @@ -165,43 +349,105 @@ def get_playlist_info(): @app.route("/api/download", methods=["POST"]) def start_download(): - data = request.json - url = data.get("url", "").strip() - format_choice = data.get("format", "video") - format_id = data.get("format_id") - title = data.get("title", "") - - if not url: + data = request.get_json(silent=True) + if not isinstance(data, dict): + return jsonify({"error": "Invalid request body"}), 400 + url = data.get("url", "") + if not isinstance(url, str) or not url.strip(): return jsonify({"error": "No URL provided"}), 400 - - job_id = uuid.uuid4().hex[:10] - jobs[job_id] = {"status": "downloading", "url": url, "title": title} - - thread = threading.Thread(target=run_download, args=(job_id, url, format_choice, format_id)) - thread.daemon = True - thread.start() - - return jsonify({"job_id": job_id}) + if not is_safe_url(url.strip()): + return jsonify({"error": "Invalid URL"}), 400 + try: + return jsonify(_get_job_service().create(data)) + except Exception as exc: + return _service_error(exc) @app.route("/api/status/") def check_status(job_id): - job = jobs.get(job_id) - if not job: - return jsonify({"error": "Job not found"}), 404 - return jsonify({ - "status": job["status"], - "error": job.get("error"), - "filename": job.get("filename"), - }) + with jobs_lock: + job = jobs.get(job_id) + if job: + return jsonify({ + "status": job["status"], + "error": job.get("error"), + "filename": job.get("filename"), + "phase": job.get("phase"), + "progress": job.get("progress"), + }) + try: + return jsonify(_get_job_service().status(job_id)) + except Exception as exc: + return _service_error(exc) @app.route("/api/file/") def download_file(job_id): - job = jobs.get(job_id) - if not job or job["status"] != "done": - return jsonify({"error": "File not ready"}), 404 - return send_file(job["file"], as_attachment=True, download_name=job["filename"]) + with jobs_lock: + job = jobs.get(job_id) + if job: + if job["status"] != "done": + return jsonify({"error": "File not ready"}), 404 + file_path = job["file"] + filename = job["filename"] + return send_file(file_path, as_attachment=True, download_name=filename) + try: + file_path, filename = _get_job_service().file_path(job_id) + return send_file(file_path, as_attachment=True, download_name=filename) + except Exception as exc: + return _service_error(exc) + + +@app.route("/api/jobs") +def list_jobs(): + try: + limit = request.args.get("limit", default=50, type=int) + cursor = request.args.get("cursor") + return jsonify(_get_job_service().list_jobs(limit, cursor)) + except Exception as exc: + return _service_error(exc) + + +@app.route("/api/jobs//resume", methods=["POST"]) +def resume_job(job_id): + try: + return jsonify(_get_job_service().resume(job_id)), 202 + except Exception as exc: + return _service_error(exc) + + +@app.route("/api/jobs//cancel", methods=["POST"]) +def cancel_job(job_id): + data = request.get_json(silent=True) + if ( + not isinstance(data, dict) + or not isinstance(data.get("attempt_no"), int) + or isinstance(data.get("attempt_no"), bool) + or data["attempt_no"] <= 0 + ): + return jsonify({"error": "Invalid attempt number"}), 400 + try: + result = _get_job_service().cancel(job_id, data["attempt_no"]) + return jsonify(result), 200 if result.get("state") == "cancelled" else 202 + except Exception as exc: + return _service_error(exc) + + +@app.route("/api/jobs//restart", methods=["POST"]) +def restart_job(job_id): + try: + return jsonify(_get_job_service().restart(job_id)), 201 + except Exception as exc: + return _service_error(exc) + + +@app.route("/api/jobs/", methods=["DELETE"]) +def delete_job(job_id): + try: + _get_job_service().delete(job_id) + return ("", 204) + except Exception as exc: + return _service_error(exc) if __name__ == "__main__": diff --git a/docker-compose.yml b/docker-compose.yml index 19404ecb..2cebe62b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,7 +4,7 @@ services: image: reclip:latest container_name: reclip ports: - - "8899:8899" + - "127.0.0.1:8899:8899" volumes: - reclip-downloads:/app/downloads restart: unless-stopped diff --git a/download_plan.py b/download_plan.py new file mode 100644 index 00000000..2eeb981b --- /dev/null +++ b/download_plan.py @@ -0,0 +1,211 @@ +"""Resolve a media choice once and build safe, resumable yt-dlp commands.""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Callable + + +FORMAT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + + +def _format_id(value: Any) -> str: + value = str(value or "") + if not FORMAT_ID_RE.fullmatch(value): + raise ValueError("Unsupported format identifier") + return value + + +def _codec_present(value: Any) -> bool: + return bool(value and value != "none") + + +def _score(media_format: dict[str, Any]) -> tuple[float, float, float]: + def number(name: str) -> float: + value = media_format.get(name) + return float(value) if isinstance(value, (int, float)) and value >= 0 else 0.0 + + return number("height"), number("tbr"), number("filesize") + + +def _select_format( + formats: list[dict[str, Any]], requested_id: str | None, predicate: Callable[[dict[str, Any]], bool] +) -> dict[str, Any]: + candidates = [item for item in formats if predicate(item)] + if requested_id: + requested_id = _format_id(requested_id) + candidates = [item for item in candidates if str(item.get("format_id")) == requested_id] + if not candidates: + raise ValueError("Requested format is unavailable") + if not candidates: + raise ValueError("No compatible format is available") + return max(candidates, key=_score) + + +def _format_snapshot(media_format: dict[str, Any]) -> dict[str, Any]: + return { + "id": _format_id(media_format.get("format_id")), + "ext": media_format.get("ext"), + "protocol": media_format.get("protocol"), + "vcodec": media_format.get("vcodec", "none"), + "acodec": media_format.get("acodec", "none"), + "filesize": media_format.get("filesize"), + "filesize_approx": media_format.get("filesize_approx"), + "etag": media_format.get("etag"), + } + + +def resolve_plan( + source_url: str, + format_choice: str, + requested_format_id: str | None, + *, + info: dict[str, Any], +) -> dict[str, Any]: + """Resolve metadata into a plan whose format IDs will not change on retry.""" + if format_choice not in {"audio", "video"}: + raise ValueError("Invalid format choice") + if not isinstance(source_url, str) or not source_url.startswith(("http://", "https://")): + raise ValueError("Invalid source URL") + formats = info.get("formats") + if not isinstance(formats, list): + raise ValueError("No formats returned") + formats = [item for item in formats if isinstance(item, dict)] + extractor = info.get("extractor_key") or info.get("extractor") + video_id = info.get("id") + if not isinstance(extractor, str) or not extractor or not isinstance(video_id, str) or not video_id: + raise ValueError("Incomplete resource identity") + + if format_choice == "audio": + audio_only = [ + item for item in formats + if _codec_present(item.get("acodec")) and not _codec_present(item.get("vcodec")) + ] + selected = _select_format( + audio_only or formats, + requested_format_id, + lambda item: _codec_present(item.get("acodec")), + ) + selected_formats = [_format_snapshot(selected)] + selector = selected_formats[0]["id"] + else: + video = _select_format( + formats, + requested_format_id, + lambda item: _codec_present(item.get("vcodec")), + ) + selected_formats = [_format_snapshot(video)] + selector = selected_formats[0]["id"] + if not _codec_present(video.get("acodec")): + audio = _select_format( + formats, + None, + lambda item: _codec_present(item.get("acodec")) and not _codec_present(item.get("vcodec")), + ) + audio_snapshot = _format_snapshot(audio) + selected_formats.append(audio_snapshot) + selector = f"{selector}+{audio_snapshot['id']}" + + return { + "source_url": source_url, + "extractor": extractor, + "video_id": video_id, + "format_choice": format_choice, + "requested_format_id": requested_format_id, + "formats": selected_formats, + "format_selector": selector, + } + + +def validate_resume_plan(stored_plan: dict[str, Any], current_plan: dict[str, Any]) -> None: + """Reject a retry when the source or the selected media streams changed.""" + identity_fields = ("extractor", "video_id", "format_choice", "format_selector") + if any(stored_plan.get(field) != current_plan.get(field) for field in identity_fields): + raise ValueError("Source or format changed; restart required") + stored_formats = stored_plan.get("formats", []) + current_formats = current_plan.get("formats", []) + fields = ("id", "ext", "protocol", "vcodec", "acodec") + if len(stored_formats) != len(current_formats) or any( + any(old.get(field) != new.get(field) for field in fields) + for old, new in zip(stored_formats, current_formats) + ): + raise ValueError("Source or format changed; restart required") + + +def build_download_command(plan: dict[str, Any], task_dir: str | Path) -> list[str]: + """Build a shell-free command using only server-owned plan values.""" + task_dir = Path(task_dir).resolve() + task_dir.mkdir(parents=True, exist_ok=True) + selector = plan.get("format_selector") + if not isinstance(selector, str): + raise ValueError("Missing fixed format selector") + selector_ids = selector.split("+") + if not selector_ids or any(not FORMAT_ID_RE.fullmatch(item) for item in selector_ids): + raise ValueError("Invalid fixed format selector") + source_url = plan.get("source_url") + if not isinstance(source_url, str) or not source_url.startswith(("http://", "https://")): + raise ValueError("Invalid source URL") + + command = [ + "yt-dlp", + "--ignore-config", + "--no-playlist", + "--continue", + "--part", + "--newline", + "--progress", + "--progress-delta", + "0.5", + "--progress-template", + "download:RECLIP_PROGRESS %(progress.{status,downloaded_bytes,total_bytes,total_bytes_estimate,speed,eta})j", + "--progress-template", + "postprocess:RECLIP_POSTPROCESS %(progress.{status,postprocessor})j", + "--print", + "after_move:RECLIP_FINAL %(filepath)j", + "--no-simulate", + "--abort-on-unavailable-fragments", + "-f", + selector, + "-o", + str(task_dir / "media.%(ext)s"), + ] + if plan.get("format_choice") == "audio": + command.extend(["-x", "--audio-format", "mp3"]) + elif plan.get("format_choice") == "video": + command.extend(["--merge-output-format", "mp4"]) + else: + raise ValueError("Invalid format choice") + command.extend(["--", source_url]) + return command + + +def validate_final_file( + task_dir: str | Path, + reported_path: str | Path, + format_choice: str, + *, + probe_runner: Callable[[Path, str], Any] | None = None, +) -> dict[str, Any]: + """Validate a post-processed file and return a task-relative path.""" + if format_choice not in {"audio", "video"}: + raise ValueError("Invalid format choice") + root = Path(task_dir).resolve() + raw_path = Path(reported_path) + candidate = raw_path if raw_path.is_absolute() else root / raw_path + candidate = candidate.resolve(strict=False) + if not candidate.is_relative_to(root) or candidate == root: + raise ValueError("Final file is outside the task directory") + if Path(reported_path).is_symlink() or candidate.is_symlink(): + raise ValueError("Final file must not be a symlink") + if candidate.suffix.lower() in {".part", ".ytdl", ".tmp", ".temp"}: + raise ValueError("Final file is still temporary") + if not candidate.is_file() or candidate.stat().st_size <= 0: + raise ValueError("Final media file is missing or empty") + if probe_runner is not None and probe_runner(candidate, format_choice) is not True: + raise ValueError("Final media file failed validation") + return { + "relative_path": str(candidate.relative_to(root)), + "filename": candidate.name, + "size_bytes": candidate.stat().st_size, + } diff --git a/download_process.py b/download_process.py new file mode 100644 index 00000000..4a67921d --- /dev/null +++ b/download_process.py @@ -0,0 +1,249 @@ +"""Bounded streaming subprocess execution for ReClip downloads.""" + +import os +import queue +import signal +import subprocess +import threading +import time + + +class ProcessCancelled(Exception): + """Raised only after a requested process-tree stop is confirmed.""" + + def __init__(self, exit_code=None): + super().__init__("Download process cancelled") + self.exit_code = exit_code + + +class ProcessStopError(RuntimeError): + """Raised when the process tree or output reader cannot be confirmed stopped.""" + + +_WINDOWS_TASKKILL_TIMEOUT = 3 + + +class DeadlineTracker: + """Track preparation, useful download progress, post-processing and hard limits.""" + + def __init__( + self, + prepare_timeout=120, + idle_timeout=300, + process_timeout=1800, + hard_timeout=21600, + clock=time.monotonic, + ): + self.prepare_timeout = prepare_timeout + self.idle_timeout = idle_timeout + self.process_timeout = process_timeout + self.hard_timeout = hard_timeout + self.clock = clock + self.started_at = clock() + self.phase = "preparing" + self.phase_started_at = self.started_at + self.last_progress_at = self.started_at + self.last_marker = None + + def observe(self, event): + now = self.clock() + if not isinstance(event, dict): + return + kind = event.get("kind") + data = event.get("data") or {} + if kind == "download": + status = data.get("status") + if status == "downloading": + if self.phase != "downloading": + self.phase = "downloading" + self.phase_started_at = now + marker = (data.get("downloaded_bytes"), data.get("fragment_index")) + if marker != self.last_marker and any(value is not None for value in marker): + self.last_marker = marker + self.last_progress_at = now + elif status == "finished": + self.phase = "processing" + self.phase_started_at = now + self.last_marker = None + elif kind == "postprocess": + if self.phase != "processing": + self.phase = "processing" + self.phase_started_at = now + + def expired_reason(self, now=None): + now = self.clock() if now is None else now + if now - self.started_at >= self.hard_timeout: + return "attempt_timeout" + if self.phase == "preparing" and now - self.phase_started_at >= self.prepare_timeout: + return "prepare_timeout" + if self.phase == "downloading" and now - self.last_progress_at >= self.idle_timeout: + return "idle_timeout" + if self.phase == "processing" and now - self.phase_started_at >= self.process_timeout: + return "process_timeout" + return None + + +def _windows_taskkill(pid, timeout): + try: + result = subprocess.run( + ["taskkill", "/PID", str(pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + return False + return result.returncode == 0 + + +def _signal_process_tree(process, process_group_id, sig, timeout): + if os.name == "nt": + return _windows_taskkill(process.pid, timeout=timeout) + try: + os.killpg(process_group_id, sig) + except ProcessLookupError: + return True + except PermissionError: + return False + return True + + +def _stop_process_and_reader(process, reader, process_group_id): + term_ok = _signal_process_tree( + process, process_group_id, signal.SIGTERM, timeout=_WINDOWS_TASKKILL_TIMEOUT + ) + try: + process.wait(timeout=2) + except subprocess.TimeoutExpired: + pass + reader.join(timeout=2) + + if process.poll() is None or reader.is_alive(): + kill_ok = _signal_process_tree( + process, + process_group_id, + getattr(signal, "SIGKILL", signal.SIGTERM), + timeout=_WINDOWS_TASKKILL_TIMEOUT, + ) + try: + process.wait(timeout=2) + except subprocess.TimeoutExpired: + pass + reader.join(timeout=2) + else: + kill_ok = True + + # The target may have exited naturally between the poll and taskkill. + # taskkill then returns a non-zero status even though there is nothing + # left to stop. The observable safety condition is that both the process + # and the pipe reader have actually terminated. + stopped = process.poll() is not None and not reader.is_alive() + return stopped + + +def run_streaming_process( + cmd, + on_line, + timeout_seconds=300, + *, + deadline_tracker=None, + cancel_event=None, +): + """Run *cmd*, forwarding output lines to *on_line* before exit. + + The process is started without a shell. The main thread owns the timeout + clock while a reader thread consumes the merged output pipe, so a silent + or stalled child cannot make timeout handling wait forever. + """ + popen_kwargs = { + "stdout": subprocess.PIPE, + "stderr": subprocess.STDOUT, + "text": True, + "encoding": "utf-8", + "errors": "replace", + "bufsize": 1, + "shell": False, + } + if os.name != "nt": + popen_kwargs["start_new_session"] = True + + if cancel_event is not None and cancel_event.is_set(): + raise ProcessCancelled() + + process = subprocess.Popen(cmd, **popen_kwargs) + if os.name == "nt": + process_group_id = None + else: + try: + process_group_id = os.getpgid(process.pid) + except ProcessLookupError: + process_group_id = process.pid + callback_errors = queue.Queue(maxsize=1) + + def consume_output(): + try: + assert process.stdout is not None + for line in process.stdout: + on_line(line.rstrip("\r\n")) + except BaseException as exc: # propagate callback and pipe failures + try: + callback_errors.put_nowait(exc) + except queue.Full: + pass + finally: + if process.stdout is not None: + process.stdout.close() + + reader = threading.Thread(target=consume_output, name="reclip-output-reader") + reader.daemon = True + reader.start() + + def stop_or_raise(): + if not _stop_process_and_reader(process, reader, process_group_id): + raise ProcessStopError("Unable to confirm download process stopped") + + deadline = time.monotonic() + timeout_seconds + try: + while process.poll() is None: + try: + callback_error = callback_errors.get_nowait() + except queue.Empty: + callback_error = None + else: + stop_or_raise() + raise callback_error + + if cancel_event is not None and cancel_event.is_set(): + stop_or_raise() + raise ProcessCancelled(process.returncode) + + remaining = deadline - time.monotonic() + if remaining <= 0: + stop_or_raise() + timeout = subprocess.TimeoutExpired(cmd, timeout_seconds) + timeout.reason = "attempt_timeout" + raise timeout + if deadline_tracker is not None: + reason = deadline_tracker.expired_reason() + if reason is not None: + stop_or_raise() + timeout = subprocess.TimeoutExpired(cmd, time.monotonic() - deadline + timeout_seconds) + timeout.reason = reason + raise timeout + time.sleep(min(0.05, remaining)) + + reader.join(timeout=2) + if reader.is_alive(): + stop_or_raise() + try: + callback_error = callback_errors.get_nowait() + except queue.Empty: + callback_error = None + else: + raise callback_error + return process.returncode + except BaseException: + if process.poll() is None or reader.is_alive(): + stop_or_raise() + raise diff --git a/job_service.py b/job_service.py new file mode 100644 index 00000000..da6fbd04 --- /dev/null +++ b/job_service.py @@ -0,0 +1,537 @@ +"""Task lifecycle orchestration for durable and resumable downloads.""" + +from __future__ import annotations + +import json +import subprocess +import threading +import time +import uuid +from pathlib import Path +from urllib.parse import urlparse +from typing import Any, Callable + +from download_plan import ( + build_download_command, + resolve_plan, + validate_final_file, + validate_resume_plan, +) +from download_process import ( + DeadlineTracker, + ProcessCancelled, + ProcessStopError, + run_streaming_process, +) +from job_store import DOWNLOAD_ACTIVE_STATES, JobStore +from progress import normalize_download_progress, parse_progress_line + + +def _empty_progress() -> dict[str, Any]: + return normalize_download_progress({}, now=None) + + +def _safe_source_url(url: Any) -> bool: + if not isinstance(url, str) or len(url) > 4096: + return False + try: + parsed = urlparse(url) + except ValueError: + return False + return ( + parsed.scheme in {"http", "https"} + and bool(parsed.hostname) + and parsed.username is None + and parsed.password is None + and not url.startswith(("-", "--")) + ) + + +def load_video_info(source_url: str) -> dict[str, Any]: + """Fetch one metadata object without invoking a shell or saving output.""" + result = subprocess.run( + ["yt-dlp", "--ignore-config", "--no-playlist", "-j", "--", source_url], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=120, + check=False, + ) + if result.returncode != 0: + raise ValueError("Unable to fetch media information") + for line in result.stdout.splitlines(): + if line.strip(): + try: + info = json.loads(line) + except ValueError: + break + if isinstance(info, dict): + return info + break + raise ValueError("Media information was empty") + + +class CancelUnavailableError(RuntimeError): + """The requested attempt is active in storage but has no local control.""" + + +class JobService: + def __init__( + self, + store: JobStore, + download_root: str | Path, + runtime_epoch: str, + *, + metadata_loader: Callable[[str], dict[str, Any]] = load_video_info, + runner: Callable[..., int] = run_streaming_process, + thread_factory: Callable[..., Any] = threading.Thread, + probe_runner: Callable[[Path, str], Any] | None = None, + prepare_timeout: int = 120, + idle_timeout: int = 300, + process_timeout: int = 1800, + max_attempt_seconds: int = 21600, + ): + self.store = store + self.download_root = Path(download_root).resolve() + self.runtime_epoch = runtime_epoch + self.metadata_loader = metadata_loader + self.runner = runner + self.thread_factory = thread_factory + self.probe_runner = probe_runner + self.prepare_timeout = prepare_timeout + self.idle_timeout = idle_timeout + self.process_timeout = process_timeout + self.max_attempt_seconds = max_attempt_seconds + self._attempt_controls: dict[tuple[str, int], threading.Event] = {} + self._control_lock = threading.RLock() + + def create(self, request_data: dict[str, Any]) -> dict[str, Any]: + if not isinstance(request_data, dict): + raise ValueError("Invalid request body") + source_url = request_data.get("url", "") + format_choice = request_data.get("format", "video") + requested_format_id = request_data.get("format_id") + if not _safe_source_url(source_url): + raise ValueError("Invalid URL") + if format_choice not in {"audio", "video"}: + raise ValueError("Invalid format choice") + if requested_format_id is not None and not isinstance(requested_format_id, str): + raise ValueError("Invalid format selection") + info = self.metadata_loader(source_url) + plan = resolve_plan( + source_url, + format_choice, + requested_format_id, + info=info, + ) + job_id = uuid.uuid4().hex + title = request_data.get("title") or info.get("title") or "" + if not isinstance(title, str): + title = "" + title = title[:500] + now = time.time() + record = { + "job_id": job_id, + "source_url": source_url, + "title": title, + "format_choice": format_choice, + "requested_format_id": requested_format_id, + "state": "preparing", + "attempt_no": 1, + "resource_json": plan, + "progress_json": _empty_progress(), + "last_progress_json": None, + "error_code": None, + "error_message": None, + "final_relpath": None, + "filename": None, + "created_at": now, + "updated_at": now, + } + self._task_dir(job_id).mkdir(parents=True, exist_ok=False) + try: + self.store.insert_job(record) + except Exception: + self._task_dir(job_id).rmdir() + raise + self._start_attempt(job_id, 1, plan) + return {"job_id": job_id, "attempt_no": 1} + + def resume(self, job_id: str) -> dict[str, Any]: + job = self._require_job(job_id) + if job["state"] not in {"failed", "interrupted", "cancelled"}: + raise RuntimeError("Job cannot be resumed in its current state") + current_plan = resolve_plan( + job["source_url"], + job["format_choice"], + job["requested_format_id"], + info=self.metadata_loader(job["source_url"]), + ) + validate_resume_plan(job["resource_json"], current_plan) + attempt_no = self.store.claim_retry(job_id, job["attempt_no"], self.runtime_epoch) + if attempt_no is None: + raise RuntimeError("Job changed before it could be resumed") + self._start_attempt(job_id, attempt_no, job["resource_json"]) + return {"job_id": job_id, "attempt_no": attempt_no} + + def cancel(self, job_id: str, attempt_no: int) -> dict[str, Any]: + if not isinstance(attempt_no, int) or isinstance(attempt_no, bool) or attempt_no <= 0: + raise ValueError("Invalid attempt number") + job = self._require_job(job_id) + state = job["state"] + if state in {"cancelled", "cancelling"} and job["attempt_no"] == attempt_no: + return self.status(job_id) + if state in {"cancelled", "cancelling"}: + raise RuntimeError("Job changed before it could be cancelled") + if state not in DOWNLOAD_ACTIVE_STATES: + raise RuntimeError("Job cannot be cancelled in its current state") + if job["attempt_no"] != attempt_no: + raise RuntimeError("Job changed before it could be cancelled") + + key = (job_id, attempt_no) + with self._control_lock: + cancel_event = self._attempt_controls.get(key) + if cancel_event is None: + raise CancelUnavailableError("Active download control is unavailable") + + if not self.store.request_cancel(job_id, attempt_no): + current = self._require_job(job_id) + if current["state"] in {"cancelling", "cancelled"}: + return self.status(job_id) + raise RuntimeError("Job changed before it could be cancelled") + + cancel_event.set() + return self.status(job_id) + + def restart(self, job_id: str) -> dict[str, Any]: + job = self._require_job(job_id) + if job["state"] in {"preparing", "downloading", "processing", "deleting"}: + raise RuntimeError("Job is still active") + return self.create( + { + "url": job["source_url"], + "format": job["format_choice"], + "format_id": job["requested_format_id"], + "title": job["title"], + } + ) + + def status(self, job_id: str) -> dict[str, Any]: + job = self._require_job(job_id) + status = { + "preparing": "downloading", + "downloading": "downloading", + "processing": "downloading", + "cancelling": "downloading", + "completed": "done", + "failed": "error", + "interrupted": "error", + "deleting": "error", + }.get(job["state"], "error") + phase = { + "preparing": "preparing", + "downloading": "downloading", + "processing": "processing", + "cancelling": "cancelling", + "cancelled": "cancelled", + "completed": "complete", + "failed": "failed", + "interrupted": "interrupted", + "deleting": "deleting", + }.get(job["state"], "failed") + return { + "status": status, + "state": job["state"], + "attempt_no": job["attempt_no"], + "error": job["error_message"], + "error_code": job["error_code"], + "filename": job["filename"], + "phase": phase, + "progress": job["progress_json"], + "last_progress": job["last_progress_json"], + "can_retry": job["state"] in {"failed", "interrupted", "cancelled"}, + "resume_candidate": job["state"] in {"failed", "interrupted", "cancelled"} + and bool(job["resource_json"]), + "resume_result": "unknown", + "job_id": job["job_id"], + "can_cancel": job["state"] in DOWNLOAD_ACTIVE_STATES, + } + + def list_jobs(self, limit: int = 50, cursor: str | None = None) -> dict[str, Any]: + result = self.store.list_jobs(limit, cursor) + result["items"] = [self._public_job(item) for item in result["items"]] + return result + + def file_path(self, job_id: str) -> tuple[Path, str]: + job = self._require_job(job_id) + if job["state"] != "completed" or not job["final_relpath"] or not job["filename"]: + raise RuntimeError("File is not ready") + root = self.download_root.resolve() + candidate = (root / Path(job["final_relpath"])).resolve(strict=False) + task_dir = self._task_dir(job_id).resolve() + if not candidate.is_relative_to(task_dir) or candidate.is_symlink() or not candidate.is_file(): + raise FileNotFoundError("File is no longer available") + return candidate, job["filename"] + + def delete(self, job_id: str) -> None: + job = self._require_job(job_id) + if job["state"] == "deleted": + return + if not self.store.begin_delete(job_id): + raise RuntimeError("Job is active or cannot be deleted") + try: + task_dir = self._task_dir(job_id) + if task_dir.exists(): + import shutil + + shutil.rmtree(task_dir) + self.store.finish_delete(job_id) + except Exception: + raise + + def recover(self) -> dict[str, Any]: + result = self.store.recover_active_jobs(self.runtime_epoch) + if result.get("restart_required"): + raise RuntimeError("Service restart required before recovering active jobs") + return result + + def run_attempt( + self, + job_id: str, + attempt_no: int, + plan: dict[str, Any], + cancel_event: threading.Event | None = None, + ) -> None: + if cancel_event is None: + with self._control_lock: + cancel_event = self._attempt_controls.get((job_id, attempt_no)) + if cancel_event is None: + cancel_event = threading.Event() + task_dir = self._task_dir(job_id) + command = build_download_command(plan, task_dir) + final_path: str | None = None + last_error_lines: list[str] = [] + last_persisted = 0.0 + deadline_tracker = DeadlineTracker( + prepare_timeout=self.prepare_timeout, + idle_timeout=self.idle_timeout, + process_timeout=self.process_timeout, + hard_timeout=self.max_attempt_seconds, + ) + + def handle_line(line: str) -> None: + nonlocal final_path, last_persisted + if cancel_event.is_set(): + return + event = parse_progress_line(line) + if event is None: + if line.startswith("ERROR:") or line.startswith("WARNING:"): + last_error_lines.append(self._clean_error(line)) + del last_error_lines[:-20] + return + deadline_tracker.observe(event) + now = time.time() + if event["kind"] == "final": + final_path = event["path"] + return + if event["kind"] == "download": + data = event["data"] + if data.get("status") == "downloading": + progress = normalize_download_progress(data, now) + if now - last_persisted >= 2.0: + updated = self.store.update_attempt( + job_id, + attempt_no, + {"state": "downloading", "progress_json": progress}, + expected_states=DOWNLOAD_ACTIVE_STATES, + ) + if updated: + last_persisted = now + elif data.get("status") == "finished": + progress = normalize_download_progress(data, now) + self.store.update_attempt( + job_id, + attempt_no, + {"state": "processing", "progress_json": None, "last_progress_json": progress}, + expected_states=DOWNLOAD_ACTIVE_STATES, + ) + last_persisted = now + elif event["kind"] == "postprocess": + self.store.update_attempt( + job_id, + attempt_no, + {"state": "processing", "progress_json": None}, + expected_states=DOWNLOAD_ACTIVE_STATES, + ) + + try: + returncode = self.runner( + command, + handle_line, + timeout_seconds=self.max_attempt_seconds, + deadline_tracker=deadline_tracker, + cancel_event=cancel_event, + ) + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, returncode) + return + if returncode != 0: + self._fail( + job_id, + attempt_no, + "downloader_failed", + last_error_lines[-1] if last_error_lines else "Download failed", + returncode, + ) + return + if final_path is None: + self._fail(job_id, attempt_no, "final_file_missing", "Download completed without a final file", returncode) + return + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, returncode) + return + result = validate_final_file( + task_dir, + final_path, + plan["format_choice"], + probe_runner=self.probe_runner, + ) + final_relpath = (Path("jobs") / job_id / result["relative_path"]).as_posix() + complete_progress = _empty_progress() + complete_progress["percent"] = 100.0 + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, returncode) + return + if not self.store.finalize_attempt( + job_id, + attempt_no, + "completed", + { + "progress_json": complete_progress, + "final_relpath": final_relpath, + "filename": result["filename"], + }, + returncode, + ) and cancel_event.is_set(): + self._cancelled(job_id, attempt_no, returncode) + except ProcessCancelled as exc: + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, exc.exit_code) + else: + self._fail(job_id, attempt_no, "download_failed", "Download failed", exc.exit_code) + except subprocess.TimeoutExpired as exc: + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, None) + else: + reason = getattr(exc, "reason", "attempt_timeout") + self._fail(job_id, attempt_no, reason, f"Download timed out ({reason})", None) + except (OSError, ValueError) as exc: + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, None) + else: + self._fail(job_id, attempt_no, "download_failed", str(exc)[:1000], None) + except ProcessStopError: + if not cancel_event.is_set(): + self._fail(job_id, attempt_no, "download_failed", "Unable to stop download process", None) + except Exception: + if cancel_event.is_set(): + self._cancelled(job_id, attempt_no, None) + else: + self._fail(job_id, attempt_no, "download_failed", "Download failed", None) + finally: + self._remove_attempt_control(job_id, attempt_no, cancel_event) + + def _start_attempt(self, job_id: str, attempt_no: int, plan: dict[str, Any]) -> None: + try: + cancel_event = threading.Event() + with self._control_lock: + self._attempt_controls[(job_id, attempt_no)] = cancel_event + thread = self.thread_factory( + target=self.run_attempt, + args=(job_id, attempt_no, plan, cancel_event), + name=f"reclip-download-{job_id}", + daemon=True, + ) + thread.start() + except Exception: + with self._control_lock: + self._attempt_controls.pop((job_id, attempt_no), None) + self._fail(job_id, attempt_no, "thread_start_failed", "Download could not be started", None) + + def _remove_attempt_control( + self, job_id: str, attempt_no: int, cancel_event: threading.Event + ) -> None: + with self._control_lock: + if self._attempt_controls.get((job_id, attempt_no)) is cancel_event: + self._attempt_controls.pop((job_id, attempt_no), None) + + def _cancelled(self, job_id: str, attempt_no: int, exit_code: int | None) -> bool: + return self.store.finalize_attempt( + job_id, + attempt_no, + "cancelled", + {"progress_json": None}, + exit_code, + ) + + def _fail( + self, + job_id: str, + attempt_no: int, + error_code: str, + message: str, + exit_code: int | None, + ) -> bool: + return self.store.finalize_attempt( + job_id, + attempt_no, + "failed", + { + "progress_json": None, + "error_code": error_code, + "error_message": message, + }, + exit_code, + ) + + def _require_job(self, job_id: str) -> dict[str, Any]: + if not isinstance(job_id, str) or not reclip_job_id(job_id): + raise KeyError("Job not found") + job = self.store.get_job(job_id) + if job is None or job["state"] == "deleted": + raise KeyError("Job not found") + return job + + def _task_dir(self, job_id: str) -> Path: + if not reclip_job_id(job_id): + raise ValueError("Invalid job ID") + return self.download_root / "jobs" / job_id + + @staticmethod + def _clean_error(line: str) -> str: + return line.split(":", 1)[1].strip()[:1000] + + @staticmethod + def _public_job(job: dict[str, Any]) -> dict[str, Any]: + return { + "job_id": job["job_id"], + "title": job["title"], + "format": job["format_choice"], + "state": job["state"], + "attempt_no": job["attempt_no"], + "created_at": job["created_at"], + "updated_at": job["updated_at"], + "filename": job["filename"], + "error": job["error_message"], + "progress": job["progress_json"], + "last_progress": job["last_progress_json"], + "can_retry": job["state"] in {"failed", "interrupted", "cancelled"}, + "can_cancel": job["state"] in DOWNLOAD_ACTIVE_STATES, + } + + +def reclip_job_id(value: Any) -> bool: + if not isinstance(value, str) or len(value) != 32: + return False + return all(char in "0123456789abcdef" for char in value) diff --git a/job_store.py b/job_store.py new file mode 100644 index 00000000..f730b125 --- /dev/null +++ b/job_store.py @@ -0,0 +1,636 @@ +"""Durable task state for resumable ReClip downloads.""" + +from __future__ import annotations + +import base64 +import json +import sqlite3 +import time +from pathlib import Path +from typing import Any + + +SCHEMA_VERSION = 2 + + +JOB_STATES = { + "preparing", + "downloading", + "processing", + "cancelling", + "cancelled", + "completed", + "failed", + "interrupted", + "deleting", + "deleted", +} + +JOB_FIELDS = { + "state", + "attempt_no", + "resource_json", + "progress_json", + "last_progress_json", + "error_code", + "error_message", + "final_relpath", + "filename", + "updated_at", +} + +ACTIVE_JOB_STATES = frozenset({"preparing", "downloading", "processing", "cancelling"}) +DOWNLOAD_ACTIVE_STATES = frozenset({"preparing", "downloading", "processing"}) + + +def _encode_json(value: Any) -> str | None: + if value is None: + return None + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _decode_json(value: str | None) -> Any: + if value is None: + return None + return json.loads(value) + + +def _encode_cursor(created_at: float, job_id: str) -> str: + payload = json.dumps( + {"created_at": created_at, "job_id": job_id}, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") + + +def _decode_cursor(cursor: str | None) -> tuple[float, str] | None: + if cursor is None: + return None + if not isinstance(cursor, str) or not cursor or len(cursor) > 256: + raise ValueError("Invalid cursor") + try: + padded = cursor + "=" * (-len(cursor) % 4) + data = json.loads(base64.urlsafe_b64decode(padded.encode("ascii"))) + created_at = float(data["created_at"]) + job_id = data["job_id"] + except (ValueError, KeyError, TypeError, json.JSONDecodeError): + raise ValueError("Invalid cursor") from None + if not isinstance(job_id, str) or len(job_id) != 32: + raise ValueError("Invalid cursor") + return created_at, job_id + + +class JobStore: + """SQLite-backed state store with one short-lived connection per operation.""" + + def __init__(self, db_path: str | Path): + self.db_path = Path(db_path) + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect( + self.db_path, + timeout=5.0, + isolation_level=None, + ) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA busy_timeout=5000") + connection.execute("PRAGMA foreign_keys=ON") + return connection + + def initialize(self) -> None: + self.db_path.parent.mkdir(parents=True, exist_ok=True) + connection = self._connect() + try: + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA synchronous=FULL") + version = connection.execute("PRAGMA user_version").fetchone()[0] + if version > SCHEMA_VERSION: + raise RuntimeError(f"Unsupported jobs database schema: {version}") + connection.execute("BEGIN IMMEDIATE") + connection.execute( + """ + CREATE TABLE IF NOT EXISTS jobs ( + job_id TEXT PRIMARY KEY, + source_url TEXT, + title TEXT, + format_choice TEXT NOT NULL, + requested_format_id TEXT, + state TEXT NOT NULL, + attempt_no INTEGER NOT NULL, + resource_json TEXT, + progress_json TEXT, + last_progress_json TEXT, + error_code TEXT, + error_message TEXT, + final_relpath TEXT, + filename TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS attempts ( + job_id TEXT NOT NULL REFERENCES jobs(job_id) ON DELETE CASCADE, + attempt_no INTEGER NOT NULL, + runtime_epoch TEXT, + started_at REAL NOT NULL, + finished_at REAL, + exit_code INTEGER, + outcome TEXT NOT NULL, + PRIMARY KEY (job_id, attempt_no) + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS service_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS idx_jobs_listing + ON jobs (state, created_at DESC, job_id DESC) + """ + ) + connection.execute(f"PRAGMA user_version={SCHEMA_VERSION}") + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def close(self) -> None: + """Keep the public lifecycle API; connections are per-operation.""" + + def insert_job(self, record: dict[str, Any]) -> None: + required = { + "job_id", + "source_url", + "title", + "format_choice", + "requested_format_id", + "state", + "attempt_no", + "resource_json", + "progress_json", + "error_code", + "error_message", + "final_relpath", + "filename", + "created_at", + "updated_at", + } + missing = required.difference(record) + if missing: + raise ValueError(f"Missing job fields: {sorted(missing)}") + if record["state"] not in JOB_STATES: + raise ValueError("Invalid job state") + + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + """ + INSERT INTO jobs ( + job_id, source_url, title, format_choice, requested_format_id, + state, attempt_no, resource_json, progress_json, + last_progress_json, error_code, error_message, final_relpath, + filename, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + record["job_id"], + record["source_url"], + record["title"], + record["format_choice"], + record["requested_format_id"], + record["state"], + record["attempt_no"], + _encode_json(record["resource_json"]), + _encode_json(record["progress_json"]), + _encode_json(record.get("last_progress_json")), + record["error_code"], + record["error_message"], + record["final_relpath"], + record["filename"], + record["created_at"], + record["updated_at"], + ), + ) + connection.execute( + """ + INSERT INTO attempts + (job_id, attempt_no, started_at, outcome) + VALUES (?, ?, ?, ?) + """, + ( + record["job_id"], + record["attempt_no"], + record["created_at"], + "running" if record["state"] in ACTIVE_JOB_STATES else record["state"], + ), + ) + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + @staticmethod + def _row_to_dict(row: sqlite3.Row | None) -> dict[str, Any] | None: + if row is None: + return None + result = dict(row) + for field in ("resource_json", "progress_json", "last_progress_json"): + result[field] = _decode_json(result[field]) + return result + + def get_job(self, job_id: str) -> dict[str, Any] | None: + connection = self._connect() + try: + row = connection.execute("SELECT * FROM jobs WHERE job_id = ?", (job_id,)).fetchone() + return self._row_to_dict(row) + finally: + connection.close() + + def list_jobs(self, limit: int = 50, cursor: str | None = None) -> dict[str, Any]: + if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100: + raise ValueError("Invalid limit") + decoded = _decode_cursor(cursor) + connection = self._connect() + try: + if decoded is None: + rows = connection.execute( + """ + SELECT * FROM jobs WHERE state != 'deleted' + ORDER BY created_at DESC, job_id DESC LIMIT ? + """, + (limit + 1,), + ).fetchall() + else: + created_at, job_id = decoded + rows = connection.execute( + """ + SELECT * FROM jobs + WHERE state != 'deleted' + AND (created_at < ? OR (created_at = ? AND job_id < ?)) + ORDER BY created_at DESC, job_id DESC LIMIT ? + """, + (created_at, created_at, job_id, limit + 1), + ).fetchall() + has_next = len(rows) > limit + rows = rows[:limit] + items = [self._row_to_dict(row) for row in rows] + next_cursor = None + if has_next and items: + last = items[-1] + next_cursor = _encode_cursor(last["created_at"], last["job_id"]) + return {"items": items, "next_cursor": next_cursor} + finally: + connection.close() + + def claim_retry( + self, + job_id: str, + expected_attempt: int, + runtime_epoch: str, + ) -> int | None: + now = time.time() + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + "SELECT state, attempt_no FROM jobs WHERE job_id = ?", (job_id,) + ).fetchone() + if ( + row is None + or row["attempt_no"] != expected_attempt + or row["state"] not in {"failed", "interrupted", "cancelled"} + ): + connection.rollback() + return None + new_attempt = expected_attempt + 1 + connection.execute( + """ + UPDATE jobs + SET state='preparing', attempt_no=?, progress_json=NULL, + error_code=NULL, error_message=NULL, updated_at=? + WHERE job_id=? AND attempt_no=? + """, + (new_attempt, now, job_id, expected_attempt), + ) + connection.execute( + """ + INSERT INTO attempts + (job_id, attempt_no, runtime_epoch, started_at, outcome) + VALUES (?, ?, ?, ?, 'running') + """, + (job_id, new_attempt, runtime_epoch, now), + ) + connection.commit() + return new_attempt + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def update_attempt( + self, + job_id: str, + attempt_no: int, + fields: dict[str, Any], + *, + expected_states: set[str] | frozenset[str] | None = None, + ) -> bool: + unknown = set(fields).difference(JOB_FIELDS) + if unknown: + raise ValueError(f"Unknown job fields: {sorted(unknown)}") + if not fields: + return False + if "state" in fields and fields["state"] not in JOB_STATES: + raise ValueError("Invalid job state") + if expected_states is not None: + expected_states = set(expected_states) + if not expected_states: + return False + if not expected_states.issubset(JOB_STATES): + raise ValueError("Invalid expected job states") + values = dict(fields) + for field in ("resource_json", "progress_json", "last_progress_json"): + if field in values: + values[field] = _encode_json(values[field]) + values.setdefault("updated_at", time.time()) + assignments = ", ".join(f"{field} = ?" for field in values) + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + params = [values[field] for field in values] + params.extend((job_id, attempt_no)) + where = "job_id=? AND attempt_no=?" + if expected_states is not None: + placeholders = ", ".join("?" for _ in expected_states) + where += f" AND state IN ({placeholders})" + params.extend(sorted(expected_states)) + cursor = connection.execute( + f"UPDATE jobs SET {assignments} WHERE {where}", + params, + ) + if cursor.rowcount != 1: + connection.rollback() + return False + connection.commit() + return True + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def request_cancel(self, job_id: str, attempt_no: int) -> bool: + if not isinstance(attempt_no, int) or isinstance(attempt_no, bool) or attempt_no <= 0: + return False + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + """ + UPDATE jobs + SET state='cancelling', + last_progress_json=COALESCE(progress_json, last_progress_json), + progress_json=NULL, error_code=NULL, error_message=NULL, + updated_at=? + WHERE job_id=? AND attempt_no=? + AND state IN ('preparing', 'downloading', 'processing') + """, + (time.time(), job_id, attempt_no), + ) + if cursor.rowcount != 1: + connection.rollback() + return False + connection.commit() + return True + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def finalize_attempt( + self, + job_id: str, + attempt_no: int, + outcome: str, + fields: dict[str, Any], + exit_code: int | None = None, + ) -> bool: + if outcome not in {"completed", "failed", "cancelled"}: + raise ValueError("Invalid attempt outcome") + unknown = set(fields).difference(JOB_FIELDS) + if unknown: + raise ValueError(f"Unknown job fields: {sorted(unknown)}") + if "state" in fields and fields["state"] != outcome: + raise ValueError("Final job state must match attempt outcome") + + now = time.time() + values = dict(fields) + values["state"] = outcome + values["updated_at"] = now + for field in ("resource_json", "progress_json", "last_progress_json"): + if field in values: + values[field] = _encode_json(values[field]) + assignments = ", ".join(f"{field} = ?" for field in values) + expected_states = ( + DOWNLOAD_ACTIVE_STATES if outcome in {"completed", "failed"} else {"cancelling"} + ) + placeholders = ", ".join("?" for _ in expected_states) + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + attempt_cursor = connection.execute( + """ + UPDATE attempts + SET finished_at=?, exit_code=?, outcome=? + WHERE job_id=? AND attempt_no=? AND outcome='running' + """, + (now, exit_code, outcome, job_id, attempt_no), + ) + if attempt_cursor.rowcount != 1: + connection.rollback() + return False + + params = [values[field] for field in values] + params.extend((job_id, attempt_no)) + params.extend(sorted(expected_states)) + job_cursor = connection.execute( + f""" + UPDATE jobs SET {assignments} + WHERE job_id=? AND attempt_no=? AND state IN ({placeholders}) + """, + params, + ) + if job_cursor.rowcount != 1: + connection.rollback() + return False + connection.commit() + return True + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def recover_active_jobs(self, runtime_epoch: str) -> dict[str, Any]: + now = time.time() + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + rows = connection.execute( + "SELECT job_id, attempt_no, state FROM jobs WHERE state IN (?, ?, ?, ?)", + tuple(sorted(ACTIVE_JOB_STATES)), + ).fetchall() + previous_epoch_row = connection.execute( + "SELECT value FROM service_meta WHERE key='runtime_epoch'" + ).fetchone() + previous_epoch = previous_epoch_row["value"] if previous_epoch_row else None + if previous_epoch == runtime_epoch and rows: + connection.rollback() + return {"recovered": 0, "restart_required": True} + + for row in rows: + if row["state"] == "cancelling": + job_cursor = connection.execute( + """ + UPDATE jobs + SET state='cancelled', progress_json=NULL, + error_code=NULL, error_message=NULL, updated_at=? + WHERE job_id=? AND attempt_no=? AND state='cancelling' + """, + (now, row["job_id"], row["attempt_no"]), + ) + attempt_cursor = connection.execute( + """ + UPDATE attempts SET finished_at=?, outcome='cancelled' + WHERE job_id=? AND attempt_no=? AND outcome='running' + """, + (now, row["job_id"], row["attempt_no"]), + ) + else: + job_cursor = connection.execute( + """ + UPDATE jobs + SET state='interrupted', last_progress_json=progress_json, + progress_json=NULL, error_code='service_restarted', + error_message='Download interrupted by service restart', updated_at=? + WHERE job_id=? AND attempt_no=? + """, + (now, row["job_id"], row["attempt_no"]), + ) + attempt_cursor = connection.execute( + """ + UPDATE attempts SET finished_at=?, outcome='interrupted' + WHERE job_id=? AND attempt_no=? AND outcome='running' + """, + (now, row["job_id"], row["attempt_no"]), + ) + if job_cursor.rowcount != 1 or attempt_cursor.rowcount != 1: + raise RuntimeError("Unable to safely recover active job") + connection.execute( + """ + INSERT INTO service_meta(key, value) VALUES('runtime_epoch', ?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value + """, + (runtime_epoch,), + ) + connection.commit() + return {"recovered": len(rows), "restart_required": False} + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def finish_attempt( + self, + job_id: str, + attempt_no: int, + outcome: str, + exit_code: int | None = None, + ) -> bool: + if outcome not in {"completed", "failed", "interrupted"}: + raise ValueError("Invalid attempt outcome") + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + """ + UPDATE attempts + SET finished_at=?, exit_code=?, outcome=? + WHERE job_id=? AND attempt_no=? AND outcome='running' + """, + (time.time(), exit_code, outcome, job_id, attempt_no), + ) + if cursor.rowcount != 1: + connection.rollback() + return False + connection.commit() + return True + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def begin_delete(self, job_id: str) -> bool: + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + """ + UPDATE jobs SET state='deleting', updated_at=? + WHERE job_id=? AND state IN ('failed','interrupted','completed','cancelled') + """, + (time.time(), job_id), + ) + if cursor.rowcount != 1: + connection.rollback() + return False + connection.commit() + return True + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def finish_delete(self, job_id: str) -> None: + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + connection.execute("DELETE FROM attempts WHERE job_id=?", (job_id,)) + connection.execute( + """ + UPDATE jobs + SET source_url=NULL, title=NULL, format_choice='deleted', + requested_format_id=NULL, resource_json=NULL, + progress_json=NULL, last_progress_json=NULL, + error_code=NULL, error_message=NULL, final_relpath=NULL, + filename=NULL, state='deleted', updated_at=? + WHERE job_id=? AND state='deleting' + """, + (time.time(), job_id), + ) + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() diff --git a/progress.py b/progress.py new file mode 100644 index 00000000..2900c9f3 --- /dev/null +++ b/progress.py @@ -0,0 +1,91 @@ +"""Structured yt-dlp progress parsing for ReClip.""" + +import json +import math + + +DOWNLOAD_PREFIX = "RECLIP_PROGRESS " +POSTPROCESS_PREFIX = "RECLIP_POSTPROCESS " +FINAL_PREFIX = "RECLIP_FINAL " +MAX_LINE_LENGTH = 16 * 1024 + + +def _finite_nonnegative(value): + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + if not math.isfinite(value) or value < 0: + return None + return value + + +def _nonnegative_integer(value): + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return None + return value + + +def parse_progress_line(line): + """Parse one allowlisted progress line, returning a small event or None.""" + if not isinstance(line, str) or len(line) > MAX_LINE_LENGTH: + return None + + if line.startswith(DOWNLOAD_PREFIX): + kind = "download" + payload = line[len(DOWNLOAD_PREFIX):] + elif line.startswith(POSTPROCESS_PREFIX): + kind = "postprocess" + payload = line[len(POSTPROCESS_PREFIX):] + elif line.startswith(FINAL_PREFIX): + payload = line[len(FINAL_PREFIX):] + try: + path = json.loads(payload) + except (TypeError, ValueError): + return None + if not isinstance(path, str) or not path: + return None + return {"kind": "final", "path": path} + else: + return None + + try: + data = json.loads(payload) + except (TypeError, ValueError): + return None + if not isinstance(data, dict): + return None + return {"kind": kind, "data": data} + + +def normalize_download_progress(data, now): + """Return a safe current-stream progress snapshot.""" + if not isinstance(data, dict): + data = {} + + downloaded = _nonnegative_integer(data.get("downloaded_bytes")) + total = _nonnegative_integer(data.get("total_bytes")) + estimate = _nonnegative_integer(data.get("total_bytes_estimate")) + + total_is_estimate = False + selected_total = total if total and total > 0 else None + if selected_total is None and estimate and estimate > 0: + selected_total = estimate + total_is_estimate = True + + percent = None + if downloaded is not None and selected_total is not None: + percent = min(100.0, max(0.0, downloaded / selected_total * 100)) + + speed = _finite_nonnegative(data.get("speed")) + eta = _finite_nonnegative(data.get("eta")) + timestamp = _finite_nonnegative(now) + + return { + "scope": "current_stream", + "percent": percent, + "downloaded_bytes": downloaded, + "total_bytes": selected_total, + "total_is_estimate": total_is_estimate, + "speed_bps": speed, + "eta_seconds": eta, + "updated_at": timestamp, + } diff --git a/runtime_guard.py b/runtime_guard.py new file mode 100644 index 00000000..e42a08c9 --- /dev/null +++ b/runtime_guard.py @@ -0,0 +1,86 @@ +"""Runtime identity and single-owner lock for the Linux container service.""" + +from __future__ import annotations + +import os +from pathlib import Path + +if os.name == "posix": + import fcntl +else: # pragma: no cover - exercised only by a Windows deployment adapter + import msvcrt + + +def _read_boot_id() -> str: + return Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii").strip() + + +def _read_pid1_starttime() -> str: + stat = Path("/proc/1/stat").read_text(encoding="ascii") + try: + _, fields = stat.rsplit(")", 1) + except ValueError: + raise OSError("invalid /proc/1/stat") from None + values = fields.strip().split() + if len(values) <= 19: + raise OSError("incomplete /proc/1/stat") + return values[19] + + +def runtime_epoch() -> str: + """Return an identity that changes after a complete Linux container restart.""" + if os.name != "posix": + raise RuntimeError("runtime identity is only supported in the Linux container") + try: + boot_id = _read_boot_id() + starttime = _read_pid1_starttime() + except (OSError, UnicodeError) as exc: + raise RuntimeError("Cannot read runtime identity") from exc + if not boot_id or not starttime: + raise RuntimeError("Cannot read runtime identity") + return f"{boot_id}:{starttime}" + + +class RuntimeGuard: + """Hold an OS file lock for the lifetime of the task-service owner.""" + + def __init__(self, lock_path: str | Path): + self.lock_path = Path(lock_path) + self._file = None + + def acquire(self) -> str: + if self._file is not None: + raise RuntimeError("Runtime guard is already acquired") + self.lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = self.lock_path.open("a+b") + handle.seek(0) + handle.write(b"0") + handle.flush() + try: + if os.name == "posix": + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + else: # pragma: no cover - Docker deployment is Linux + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + except (BlockingIOError, OSError) as exc: + handle.close() + raise RuntimeError("Another task service already owns the runtime lock") from exc + self._file = handle + try: + return runtime_epoch() + except Exception: + self.close() + raise + + def close(self) -> None: + handle, self._file = self._file, None + if handle is None: + return + try: + if os.name == "posix": + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + else: # pragma: no cover - Docker deployment is Linux + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + finally: + handle.close() diff --git a/static/jobs.js b/static/jobs.js new file mode 100644 index 00000000..db55149e --- /dev/null +++ b/static/jobs.js @@ -0,0 +1,85 @@ +(function (root, factory) { + const api = factory(); + if (typeof module === 'object' && module.exports) { + module.exports = api; + } else { + root.ReclipJobs = api; + } +}(typeof self !== 'undefined' ? self : this, function () { + 'use strict'; + + function newerJob(previous, incoming) { + const previousAttempt = Number(previous.attempt_no || 0); + const incomingAttempt = Number(incoming.attempt_no || 0); + if (incomingAttempt !== previousAttempt) return incomingAttempt > previousAttempt; + return Number(incoming.updated_at || 0) >= Number(previous.updated_at || 0); + } + + function mergeJobs(existing, incoming) { + const result = Array.isArray(existing) ? existing.slice() : []; + const positions = new Map(result.map((job, index) => [job.job_id, index])); + for (const job of (Array.isArray(incoming) ? incoming : [])) { + if (!job || typeof job.job_id !== 'string') continue; + const position = positions.get(job.job_id); + if (position === undefined) { + positions.set(job.job_id, result.length); + result.push(job); + } else if (newerJob(result[position], job)) { + result[position] = job; + } + } + return result; + } + + function shouldApplyResponse(current, incoming, generation) { + const currentGeneration = current && + (current.generation === undefined ? current.pollGeneration : current.generation); + return Boolean( + current && currentGeneration === generation && incoming && + Number(incoming.attempt_no) === Number(current.attemptNo) + ); + } + + function shouldAutoSave(job, pageStartedJobs) { + return Boolean( + job && job.state === 'completed' && pageStartedJobs && + pageStartedJobs.has(job.job_id) + ); + } + + function jobToCard(job) { + const active = ['preparing', 'downloading', 'processing'].includes(job.state); + const cancelling = job.state === 'cancelling'; + const cancelled = job.state === 'cancelled'; + return { + jobId: job.job_id, + attemptNo: job.attempt_no, + url: '', + title: job.title || '', + format: job.format || 'video', + status: job.state === 'completed' + ? 'done' + : active + ? 'downloading' + : cancelling + ? 'cancelling' + : cancelled + ? 'cancelled' + : 'error', + phase: active || cancelling || cancelled + ? job.state + : job.state === 'interrupted' ? 'interrupted' : 'failed', + progress: job.progress || null, + lastProgress: job.last_progress || null, + filename: job.filename || '', + error: job.error || '', + canRetry: job.can_retry === undefined + ? ['failed', 'interrupted', 'cancelled'].includes(job.state) + : Boolean(job.can_retry), + canCancel: job.can_cancel === undefined ? active : Boolean(job.can_cancel), + restored: true, + }; + } + + return { mergeJobs, shouldApplyResponse, shouldAutoSave, jobToCard }; +})); diff --git a/static/progress.js b/static/progress.js new file mode 100644 index 00000000..d1a2dd1f --- /dev/null +++ b/static/progress.js @@ -0,0 +1,92 @@ +(function (root, factory) { + const api = factory(); + if (typeof module === 'object' && module.exports) { + module.exports = api; + } else { + root.ReclipProgress = api; + } +}(typeof self !== 'undefined' ? self : this, function () { + 'use strict'; + + const BYTE_UNITS = ['B', 'KiB', 'MiB', 'GiB', 'TiB']; + const PHASE_TEXT = { + preparing: 'Preparing download', + downloading: 'Downloading current stream', + finalizing: 'Finishing current stream', + processing: 'Processing media', + cancelling: 'Cancelling download', + cancelled: 'Download cancelled', + complete: 'Ready to save', + failed: 'Download failed', + }; + + function finiteNonnegative(value) { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 + ? value : null; + } + + function formatBytes(value) { + value = finiteNonnegative(value); + if (value === null) return '—'; + let unit = 0; + let amount = value; + while (amount >= 1024 && unit < BYTE_UNITS.length - 1) { + amount /= 1024; + unit += 1; + } + const decimals = unit === 0 || amount >= 100 ? 0 : amount >= 10 ? 1 : 2; + return `${amount.toFixed(decimals)} ${BYTE_UNITS[unit]}`; + } + + function formatSpeed(value) { + const formatted = formatBytes(value); + return formatted === '—' ? formatted : `${formatted}/s`; + } + + function formatEta(value) { + value = finiteNonnegative(value); + if (value === null) return '—'; + let seconds = Math.ceil(value); + const hours = Math.floor(seconds / 3600); + seconds %= 3600; + const minutes = Math.floor(seconds / 60); + seconds %= 60; + const mm = String(minutes).padStart(2, '0'); + const ss = String(seconds).padStart(2, '0'); + return hours > 0 ? `${hours}:${mm}:${ss}` : `${mm}:${ss}`; + } + + function formatSize(progress) { + if (!progress || progress.downloaded_bytes === null || progress.downloaded_bytes === undefined) { + return '—'; + } + const downloaded = formatBytes(progress.downloaded_bytes); + if (progress.total_bytes === null || progress.total_bytes === undefined) { + return `${downloaded} downloaded`; + } + const estimate = progress.total_is_estimate ? '~' : ''; + return `${downloaded} / ${estimate}${formatBytes(progress.total_bytes)}`; + } + + function progressViewModel(job, nowMs) { + const progress = job && job.progress ? job.progress : {}; + const rawPercent = finiteNonnegative(progress.percent); + const percent = rawPercent === null ? null : Math.min(100, rawPercent); + const updatedAt = finiteNonnegative(progress.updated_at); + const stale = updatedAt !== null && Number.isFinite(nowMs) + ? (nowMs / 1000) - updatedAt > 5 : false; + const phase = job && job.phase ? job.phase : 'preparing'; + return { + phaseText: PHASE_TEXT[phase] || 'Downloading', + percent, + sizeText: formatSize(progress), + speedText: stale ? '—' : formatSpeed(progress.speed_bps), + etaText: stale ? '—' : formatEta(progress.eta_seconds), + stale, + indeterminate: percent === null, + waitingForUpdate: stale, + }; + } + + return { formatBytes, formatSpeed, formatEta, progressViewModel }; +})); diff --git a/templates/index.html b/templates/index.html index 1f566d14..0729905b 100644 --- a/templates/index.html +++ b/templates/index.html @@ -261,15 +261,51 @@ .card-dl-btn:disabled { opacity: 0.4; cursor: not-allowed; } .card-dl-btn.done { background: var(--success); color: #fff; } .card-dl-btn.error { background: var(--error); color: #fff; } + .card-dl-btn.cancel { background: transparent; color: var(--error); border: 1px solid rgba(196, 61, 61, 0.35); } + .card-dl-btn.cancel:hover { background: rgba(196, 61, 61, 0.08); } .card-status { font-size: 0.68rem; color: var(--muted); } .card-status.downloading { color: var(--accent); } + .card-status.cancelling { color: var(--accent); } + .card-status.cancelled { color: var(--muted); } .card-status.done { color: var(--success); } .card-status.error { color: var(--error); } + .progress-bar-wrap { + width: 100%; + height: 4px; + margin-top: 8px; + overflow: hidden; + border-radius: 2px; + background: var(--card-border); + } + .progress-bar-fill { + height: 100%; + border-radius: 2px; + background: var(--accent); + transition: width 0.3s ease; + } + .progress-details { + display: flex; + flex-wrap: wrap; + gap: 4px 10px; + margin-top: 5px; + color: var(--muted); + font-family: 'DM Mono', monospace; + font-size: 0.62rem; + } + .progress-bar-fill.indeterminate { + width: 35%; + animation: progress-slide 1.2s ease-in-out infinite; + } + @keyframes progress-slide { + from { transform: translateX(-100%); } + to { transform: translateX(285%); } + } + .q-chip { padding: 4px 10px; border: 1.5px solid var(--card-border); @@ -407,9 +443,12 @@

ReClip

+ + diff --git a/tests/fixtures/fake_downloader.py b/tests/fixtures/fake_downloader.py new file mode 100644 index 00000000..18868005 --- /dev/null +++ b/tests/fixtures/fake_downloader.py @@ -0,0 +1,38 @@ +import subprocess +import sys +import time + + +mode = sys.argv[1] + +if mode == "normal": + print( + 'RECLIP_PROGRESS {"status":"downloading","downloaded_bytes":25,"total_bytes":100}', + flush=True, + ) + time.sleep(0.05) + print( + 'RECLIP_PROGRESS {"status":"finished","downloaded_bytes":100,"total_bytes":100}', + flush=True, + ) +elif mode == "silent": + time.sleep(30) +elif mode == "fail": + print("ERROR: controlled test failure", file=sys.stderr, flush=True) + sys.exit(2) +elif mode == "flood": + for index in range(10_000): + print(f"diagnostic line {index}", flush=True) +elif mode == "spawn-child": + marker = sys.argv[2] + subprocess.Popen( + [ + sys.executable, + "-c", + "import pathlib, sys, time; time.sleep(30); pathlib.Path(sys.argv[1]).write_text('alive')", + marker, + ] + ) + time.sleep(30) +else: + raise SystemExit(f"unknown test mode: {mode}") diff --git a/tests/jobs.test.cjs b/tests/jobs.test.cjs new file mode 100644 index 00000000..d8065317 --- /dev/null +++ b/tests/jobs.test.cjs @@ -0,0 +1,72 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const jobs = require('../static/jobs.js'); + +test('mergeJobs deduplicates by job id and keeps newest attempt', () => { + const merged = jobs.mergeJobs( + [{ job_id: 'a', attempt_no: 1, state: 'failed' }], + [ + { job_id: 'a', attempt_no: 2, state: 'downloading' }, + { job_id: 'b', attempt_no: 1, state: 'completed' }, + ] + ); + assert.deepEqual(merged.map((job) => job.job_id), ['a', 'b']); + assert.equal(merged[0].attempt_no, 2); +}); + +test('stale response is rejected after resume starts', () => { + assert.equal(jobs.shouldApplyResponse({ attemptNo: 2, generation: 4 }, { attempt_no: 1 }, 4), false); + assert.equal(jobs.shouldApplyResponse({ attemptNo: 2, generation: 4 }, { attempt_no: 2 }, 4), true); + assert.equal(jobs.shouldApplyResponse({ attemptNo: 2, generation: 4 }, { attempt_no: 2 }, 3), false); +}); + +test('restored completed task never auto-saves', () => { + assert.equal(jobs.shouldAutoSave({ job_id: 'a', state: 'completed' }, new Set()), false); + assert.equal(jobs.shouldAutoSave({ job_id: 'a', state: 'completed' }, new Set(['a'])), true); +}); + +test('job card maps interrupted state to retryable error', () => { + const card = jobs.jobToCard({ + job_id: 'a', + title: 'Sample', + format: 'video', + state: 'interrupted', + attempt_no: 3, + error: 'Download interrupted', + progress: null, + last_progress: { percent: 42 }, + }); + assert.equal(card.status, 'error'); + assert.equal(card.jobId, 'a'); + assert.equal(card.canRetry, true); +}); + +test('job card maps cancelling state to a non-retryable active card', () => { + const card = jobs.jobToCard({ + job_id: 'a', + title: 'Sample', + format: 'video', + state: 'cancelling', + attempt_no: 1, + progress: null, + }); + assert.equal(card.status, 'cancelling'); + assert.equal(card.phase, 'cancelling'); + assert.equal(card.canCancel, false); + assert.equal(card.canRetry, false); +}); + +test('job card maps cancelled state to a retryable cancelled card', () => { + const card = jobs.jobToCard({ + job_id: 'a', + title: 'Sample', + format: 'video', + state: 'cancelled', + attempt_no: 1, + last_progress: { percent: 42 }, + }); + assert.equal(card.status, 'cancelled'); + assert.equal(card.phase, 'cancelled'); + assert.equal(card.canCancel, false); + assert.equal(card.canRetry, true); +}); diff --git a/tests/progress.test.cjs b/tests/progress.test.cjs new file mode 100644 index 00000000..4e66cef6 --- /dev/null +++ b/tests/progress.test.cjs @@ -0,0 +1,61 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { + formatBytes, + formatSpeed, + formatEta, + progressViewModel, +} = require('../static/progress.js'); + + +test('unknown and zero have different meanings', () => { + assert.equal(formatBytes(null), '—'); + assert.equal(formatBytes(0), '0 B'); + assert.equal(formatSpeed(0), '0 B/s'); + assert.equal(formatEta(null), '—'); + assert.equal(formatEta(65), '01:05'); + assert.equal(formatEta(3665), '1:01:05'); +}); + + +test('unknown percent remains indeterminate', () => { + const view = progressViewModel({ + status: 'downloading', + phase: 'downloading', + progress: { + percent: null, + downloaded_bytes: 1024, + total_bytes: null, + total_is_estimate: false, + speed_bps: null, + eta_seconds: null, + updated_at: 100, + }, + }, 100000); + assert.equal(view.indeterminate, true); + assert.equal(view.percent, null); + assert.equal(view.speedText, '—'); + assert.equal(view.etaText, '—'); +}); + + +test('stale progress hides speed and eta without changing the job phase', () => { + const view = progressViewModel({ + status: 'downloading', + phase: 'downloading', + progress: { + percent: 50, + downloaded_bytes: 50, + total_bytes: 100, + total_is_estimate: false, + speed_bps: 10, + eta_seconds: 5, + updated_at: 10, + }, + }, 16000); + assert.equal(view.phaseText, 'Downloading current stream'); + assert.equal(view.percent, 50); + assert.equal(view.stale, true); + assert.equal(view.speedText, '—'); + assert.equal(view.etaText, '—'); +}); diff --git a/tests/test_app_progress.py b/tests/test_app_progress.py new file mode 100644 index 00000000..7e5e657e --- /dev/null +++ b/tests/test_app_progress.py @@ -0,0 +1,195 @@ +import unittest +import uuid +from pathlib import Path +import tempfile +from unittest.mock import patch + +import app as app_module + + +class ApiProgressTests(unittest.TestCase): + def setUp(self): + with app_module.jobs_lock: + app_module.jobs.clear() + self.client = app_module.app.test_client() + + def _create_job(self): + job_id = uuid.uuid4().hex[:10] + with app_module.jobs_lock: + app_module.jobs[job_id] = { + "status": "downloading", + "phase": "preparing", + "progress": app_module._empty_progress(), + "url": "https://example.com/video.mp4", + "title": "", + } + return job_id + + def test_initial_status_has_compatible_progress_fields(self): + job_id = self._create_job() + response = self.client.get(f"/api/status/{job_id}") + payload = response.get_json() + self.assertEqual(payload["status"], "downloading") + self.assertEqual(payload["phase"], "preparing") + self.assertEqual(payload["progress"]["scope"], "current_stream") + self.assertIsNone(payload["progress"]["percent"]) + + def test_finished_stream_is_not_done(self): + job_id = self._create_job() + app_module.apply_progress_event( + job_id, + { + "kind": "download", + "data": { + "status": "finished", + "downloaded_bytes": 100, + "total_bytes": 100, + "speed": 10, + "eta": 0, + }, + }, + now=123.0, + ) + payload = self.client.get(f"/api/status/{job_id}").get_json() + self.assertEqual(payload["status"], "downloading") + self.assertEqual(payload["phase"], "finalizing") + self.assertIsNone(payload["progress"]["speed_bps"]) + self.assertIsNone(payload["progress"]["eta_seconds"]) + + def test_next_stream_replaces_previous_stream_progress(self): + job_id = self._create_job() + app_module.apply_progress_event( + job_id, + { + "kind": "download", + "data": { + "status": "finished", + "downloaded_bytes": 1000, + "total_bytes": 1000, + }, + }, + now=123.0, + ) + app_module.apply_progress_event( + job_id, + { + "kind": "download", + "data": { + "status": "downloading", + "downloaded_bytes": 10, + "total_bytes": 100, + "speed": 5, + "eta": 18, + }, + }, + now=124.0, + ) + progress = self.client.get(f"/api/status/{job_id}").get_json()["progress"] + self.assertEqual(progress["downloaded_bytes"], 10) + self.assertEqual(progress["total_bytes"], 100) + self.assertEqual(progress["percent"], 10.0) + + def test_terminal_job_ignores_late_progress(self): + job_id = self._create_job() + with app_module.jobs_lock: + app_module.jobs[job_id].update({"status": "done", "phase": "complete"}) + app_module.apply_progress_event( + job_id, + { + "kind": "download", + "data": { + "status": "downloading", + "downloaded_bytes": 50, + "total_bytes": 100, + }, + }, + now=125.0, + ) + payload = self.client.get(f"/api/status/{job_id}").get_json() + self.assertEqual(payload["status"], "done") + self.assertEqual(payload["phase"], "complete") + + def test_unsafe_url_is_rejected_before_thread_creation(self): + with patch.object(app_module.threading.Thread, "start") as start: + response = self.client.post( + "/api/download", + json={"url": "--exec=echo unsafe", "format": "video"}, + ) + self.assertEqual(response.status_code, 400) + self.assertEqual(response.get_json()["error"], "Invalid URL") + start.assert_not_called() + + def test_download_stream_updates_and_finishes_atomically(self): + job_id = self._create_job() + with tempfile.TemporaryDirectory() as directory: + previous_directory = app_module.DOWNLOAD_DIR + app_module.DOWNLOAD_DIR = directory + try: + def fake_runner(cmd, on_line, timeout_seconds): + self.assertIn("--progress-template", cmd) + self.assertIn("--", cmd) + on_line( + 'RECLIP_PROGRESS {"status":"downloading",' + '"downloaded_bytes":50,"total_bytes":100,"speed":20,"eta":3}' + ) + on_line( + 'RECLIP_POSTPROCESS {"status":"started",' + '"postprocessor":"FFmpeg"}' + ) + Path(directory, f"{job_id}.mp4").write_bytes(b"test") + return 0 + + with patch.object(app_module, "run_streaming_process", fake_runner): + app_module.run_download( + job_id, + "https://example.com/video.mp4", + "video", + None, + ) + finally: + app_module.DOWNLOAD_DIR = previous_directory + + payload = self.client.get(f"/api/status/{job_id}").get_json() + self.assertEqual(payload["status"], "done") + self.assertEqual(payload["phase"], "complete") + self.assertEqual(payload["filename"], f"{job_id}.mp4") + self.assertEqual(payload["progress"]["percent"], 100.0) + + def test_download_failure_exposes_clean_summary(self): + job_id = self._create_job() + + def fake_runner(cmd, on_line, timeout_seconds): + on_line("ERROR: private diagnostic details") + return 2 + + with patch.object(app_module, "run_streaming_process", fake_runner): + app_module.run_download( + job_id, + "https://example.com/video.mp4", + "video", + None, + ) + + payload = self.client.get(f"/api/status/{job_id}").get_json() + self.assertEqual(payload["status"], "error") + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["error"], "private diagnostic details") + self.assertIsNone(payload["progress"]) + + def test_progress_asset_is_served(self): + response = self.client.get("/static/progress.js") + try: + self.assertEqual(response.status_code, 200) + self.assertIn(b"progressViewModel", response.data) + finally: + response.close() + + def test_index_references_progress_asset_and_accessible_progress_bar(self): + response = self.client.get("/") + self.assertEqual(response.status_code, 200) + self.assertIn(b"/static/progress.js", response.data) + self.assertIn(b"role=\"progressbar\"", response.data) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_download_plan.py b/tests/test_download_plan.py new file mode 100644 index 00000000..f8b74371 --- /dev/null +++ b/tests/test_download_plan.py @@ -0,0 +1,109 @@ +import tempfile +import unittest +from pathlib import Path + +from download_plan import ( + build_download_command, + resolve_plan, + validate_final_file, + validate_resume_plan, +) + + +URL = "https://example.com/sample" + + +def sample_info(): + return { + "id": "sample-id", + "extractor_key": "generic", + "formats": [ + { + "format_id": "18", + "ext": "mp4", + "protocol": "https", + "height": 360, + "tbr": 500, + "vcodec": "avc1", + "acodec": "mp4a", + "filesize": 100, + }, + { + "format_id": "137", + "ext": "mp4", + "protocol": "https", + "height": 1080, + "tbr": 5000, + "vcodec": "avc1", + "acodec": "none", + "filesize": 1000, + }, + { + "format_id": "140", + "ext": "m4a", + "protocol": "https", + "tbr": 130, + "vcodec": "none", + "acodec": "mp4a.40.2", + "filesize": 200, + }, + ], + } + + +class DownloadPlanTests(unittest.TestCase): + def test_resolve_plan_pins_video_and_audio_ids(self): + plan = resolve_plan(URL, "video", None, info=sample_info()) + self.assertEqual(plan["video_id"], "sample-id") + self.assertEqual(plan["format_selector"], "137+140") + self.assertEqual([item["id"] for item in plan["formats"]], ["137", "140"]) + + def test_audio_plan_pins_audio_format(self): + plan = resolve_plan(URL, "audio", None, info=sample_info()) + self.assertEqual(plan["format_selector"], "140") + self.assertEqual(plan["formats"][0]["acodec"], "mp4a.40.2") + + def test_command_uses_fixed_selector_and_argument_separator(self): + plan = resolve_plan(URL, "video", None, info=sample_info()) + with tempfile.TemporaryDirectory() as directory: + command = build_download_command(plan, Path(directory)) + self.assertIn("137+140", command) + self.assertEqual(command[-2:], ["--", URL]) + self.assertIn("--continue", command) + self.assertIn("--part", command) + self.assertNotIn(URL, command[:-2]) + + def test_changed_format_cannot_resume(self): + stored = resolve_plan(URL, "video", None, info=sample_info()) + changed_info = sample_info() + changed_info["formats"][1] = { + **changed_info["formats"][1], + "format_id": "248", + } + current = resolve_plan(URL, "video", None, info=changed_info) + with self.assertRaisesRegex(ValueError, "changed"): + validate_resume_plan(stored, current) + + def test_final_path_cannot_escape_task(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) / "task" + root.mkdir() + outside = Path(directory) / "outside.mp4" + outside.write_bytes(b"media") + with self.assertRaises(ValueError): + validate_final_file(root, outside, "video") + + def test_final_path_returns_task_relative_file(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) / "task" + root.mkdir() + output = root / "media.mp4" + output.write_bytes(b"media") + result = validate_final_file(root, output, "video") + self.assertEqual(result["relative_path"], "media.mp4") + self.assertEqual(result["filename"], "media.mp4") + self.assertEqual(result["size_bytes"], 5) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_download_process.py b/tests/test_download_process.py new file mode 100644 index 00000000..6128b540 --- /dev/null +++ b/tests/test_download_process.py @@ -0,0 +1,160 @@ +import pathlib +import threading +import subprocess +import sys +import tempfile +import time +import unittest + +from download_process import ( + DeadlineTracker, + ProcessCancelled, + run_streaming_process, +) + + +FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "fake_downloader.py" + + +class ProcessTests(unittest.TestCase): + def test_deadline_tracker_detects_download_idle_timeout(self): + now = [0.0] + tracker = DeadlineTracker( + prepare_timeout=10, + idle_timeout=5, + process_timeout=20, + hard_timeout=100, + clock=lambda: now[0], + ) + self.assertIsNone(tracker.expired_reason()) + tracker.observe({"kind": "download", "data": {"status": "downloading", "downloaded_bytes": 10}}) + now[0] = 4.9 + self.assertIsNone(tracker.expired_reason()) + now[0] = 5.1 + self.assertEqual(tracker.expired_reason(), "idle_timeout") + + def test_deadline_tracker_enters_processing_after_stream_finishes(self): + now = [0.0] + tracker = DeadlineTracker( + prepare_timeout=10, + idle_timeout=5, + process_timeout=20, + hard_timeout=100, + clock=lambda: now[0], + ) + tracker.observe({"kind": "download", "data": {"status": "downloading", "downloaded_bytes": 10}}) + tracker.observe({"kind": "download", "data": {"status": "finished", "downloaded_bytes": 100}}) + now[0] = 19.9 + self.assertIsNone(tracker.expired_reason()) + now[0] = 20.1 + self.assertEqual(tracker.expired_reason(), "process_timeout") + + def test_streams_lines_before_process_exits(self): + lines = [] + started = time.monotonic() + code = run_streaming_process( + [sys.executable, str(FIXTURE), "normal"], + lines.append, + timeout_seconds=1, + ) + elapsed = time.monotonic() - started + self.assertEqual(code, 0) + self.assertEqual(len(lines), 2) + self.assertLess(elapsed, 1) + + def test_silent_process_times_out(self): + started = time.monotonic() + with self.assertRaises(subprocess.TimeoutExpired): + run_streaming_process( + [sys.executable, str(FIXTURE), "silent"], + lambda line: None, + timeout_seconds=0.2, + ) + self.assertLess(time.monotonic() - started, 5) + + def test_returns_nonzero_exit_code_and_keeps_error_line(self): + lines = [] + code = run_streaming_process( + [sys.executable, str(FIXTURE), "fail"], + lines.append, + timeout_seconds=1, + ) + self.assertEqual(code, 2) + self.assertIn("ERROR: controlled test failure", lines) + + def test_callback_error_terminates_the_process(self): + def fail_callback(line): + raise RuntimeError("callback failed") + + with self.assertRaisesRegex(RuntimeError, "callback failed"): + run_streaming_process( + [sys.executable, str(FIXTURE), "normal"], + fail_callback, + timeout_seconds=1, + ) + + def test_large_output_does_not_deadlock(self): + lines = [] + code = run_streaming_process( + [sys.executable, str(FIXTURE), "flood"], + lines.append, + timeout_seconds=5, + ) + self.assertEqual(code, 0) + self.assertEqual(len(lines), 10_000) + + def test_timeout_terminates_a_child_process(self): + with tempfile.TemporaryDirectory() as directory: + marker = pathlib.Path(directory) / "child-finished.txt" + with self.assertRaises(subprocess.TimeoutExpired): + run_streaming_process( + [sys.executable, str(FIXTURE), "spawn-child", str(marker)], + lambda line: None, + timeout_seconds=0.2, + ) + time.sleep(0.3) + self.assertFalse(marker.exists()) + + def test_precancelled_attempt_does_not_spawn(self): + from unittest.mock import patch + + cancel_event = threading.Event() + cancel_event.set() + with patch("download_process.subprocess.Popen") as popen: + with self.assertRaises(ProcessCancelled): + run_streaming_process( + [sys.executable, str(FIXTURE), "silent"], + lambda line: None, + cancel_event=cancel_event, + ) + popen.assert_not_called() + + def test_cancel_event_stops_a_silent_process(self): + cancel_event = threading.Event() + threading.Timer(0.1, cancel_event.set).start() + with self.assertRaises(ProcessCancelled): + run_streaming_process( + [sys.executable, str(FIXTURE), "silent"], + lambda line: None, + timeout_seconds=5, + cancel_event=cancel_event, + ) + + def test_cancel_event_stops_a_child_process(self): + with tempfile.TemporaryDirectory() as directory: + marker = pathlib.Path(directory) / "child-finished.txt" + cancel_event = threading.Event() + threading.Timer(0.1, cancel_event.set).start() + with self.assertRaises(ProcessCancelled): + run_streaming_process( + [sys.executable, str(FIXTURE), "spawn-child", str(marker)], + lambda line: None, + timeout_seconds=5, + cancel_event=cancel_event, + ) + time.sleep(0.3) + self.assertFalse(marker.exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_job_service.py b/tests/test_job_service.py new file mode 100644 index 00000000..fc696f87 --- /dev/null +++ b/tests/test_job_service.py @@ -0,0 +1,192 @@ +import json +import tempfile +import threading +import time +import unittest +from pathlib import Path + +from download_process import ProcessCancelled +from job_service import JobService +from job_store import JobStore + + +def sample_info(): + return { + "id": "sample-id", + "extractor_key": "generic", + "title": "Sample title", + "formats": [ + { + "format_id": "137", + "ext": "mp4", + "protocol": "https", + "height": 1080, + "tbr": 5000, + "vcodec": "avc1", + "acodec": "none", + }, + { + "format_id": "140", + "ext": "m4a", + "protocol": "https", + "tbr": 130, + "vcodec": "none", + "acodec": "mp4a", + }, + ], + } + + +class ImmediateThread: + def __init__(self, target, args, **kwargs): + self.target = target + self.args = args + + def start(self): + self.target(*self.args) + + +class FakeRunner: + def __init__(self): + self.mode = "fail" + self.calls = [] + self.kwargs = [] + + def __call__(self, command, on_line, timeout_seconds, **kwargs): + self.calls.append(command) + self.kwargs.append(kwargs) + if self.mode == "fail": + on_line("ERROR: controlled network interruption") + return 2 + output_template = Path(command[command.index("-o") + 1]) + output = Path(str(output_template).replace("%(ext)s", "mp4")) + output.write_bytes(b"completed media") + on_line( + 'RECLIP_PROGRESS ' + + json.dumps( + { + "status": "downloading", + "downloaded_bytes": 50, + "total_bytes": 100, + "speed": 20, + "eta": 3, + } + ) + ) + on_line("RECLIP_FINAL " + json.dumps(str(output))) + return 0 + + +class BlockingRunner: + def __init__(self): + self.started = threading.Event() + + def __call__(self, command, on_line, timeout_seconds, **kwargs): + cancel_event = kwargs["cancel_event"] + self.started.set() + while not cancel_event.is_set(): + time.sleep(0.01) + raise ProcessCancelled() + + +class JobServiceTests(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.root = Path(self.tempdir.name) / "downloads" + self.store = JobStore(Path(self.tempdir.name) / "jobs.sqlite3") + self.store.initialize() + self.runner = FakeRunner() + self.service = JobService( + self.store, + self.root, + runtime_epoch="epoch-A", + metadata_loader=lambda url: sample_info(), + runner=self.runner, + thread_factory=ImmediateThread, + ) + + def tearDown(self): + self.store.close() + self.tempdir.cleanup() + + def create_request(self): + return { + "url": "https://example.com/sample", + "format": "video", + "format_id": None, + "title": "Sample title", + } + + def test_failed_task_can_resume_in_same_directory(self): + created = self.service.create(self.create_request()) + failed = self.store.get_job(created["job_id"]) + task_dir = self.root / "jobs" / created["job_id"] + self.assertEqual(failed["state"], "failed") + + self.runner.mode = "success" + resumed = self.service.resume(created["job_id"]) + + self.assertEqual(resumed["job_id"], created["job_id"]) + self.assertEqual(resumed["attempt_no"], 2) + completed = self.store.get_job(created["job_id"]) + self.assertEqual(completed["state"], "completed") + self.assertTrue(task_dir.joinpath("media.mp4").exists()) + self.assertEqual(len(self.runner.calls), 2) + self.assertEqual( + self.runner.calls[0][self.runner.calls[0].index("-o") + 1], + self.runner.calls[1][self.runner.calls[1].index("-o") + 1], + ) + file_path, filename = self.service.file_path(created["job_id"]) + self.assertEqual(file_path, task_dir / "media.mp4") + self.assertEqual(filename, "media.mp4") + self.assertIsNotNone(self.runner.kwargs[1]["deadline_tracker"]) + + def test_restart_creates_new_task_and_keeps_failed_task(self): + created = self.service.create(self.create_request()) + restarted = self.service.restart(created["job_id"]) + self.assertNotEqual(restarted["job_id"], created["job_id"]) + self.assertEqual(self.store.get_job(created["job_id"])["state"], "failed") + self.assertEqual(self.store.get_job(restarted["job_id"])["state"], "failed") + + def test_resume_rejects_changed_source_plan(self): + created = self.service.create(self.create_request()) + + def changed_info(url): + data = sample_info() + data["id"] = "changed-id" + return data + + self.service.metadata_loader = changed_info + with self.assertRaisesRegex(ValueError, "changed"): + self.service.resume(created["job_id"]) + self.assertEqual(self.store.get_job(created["job_id"])["state"], "failed") + + def test_cancel_stops_current_attempt_and_marks_it_cancelled(self): + runner = BlockingRunner() + service = JobService( + self.store, + self.root, + runtime_epoch="epoch-A", + metadata_loader=lambda url: sample_info(), + runner=runner, + thread_factory=threading.Thread, + ) + + created = service.create(self.create_request()) + self.assertTrue(runner.started.wait(timeout=1)) + + response = service.cancel(created["job_id"], 1) + self.assertIn(response["state"], {"cancelling", "cancelled"}) + + deadline = time.monotonic() + 2 + while time.monotonic() < deadline: + job = self.store.get_job(created["job_id"]) + if job["state"] == "cancelled": + break + time.sleep(0.02) + self.assertEqual(self.store.get_job(created["job_id"])["state"], "cancelled") + self.assertEqual(self.store.get_job(created["job_id"])["attempt_no"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_job_store.py b/tests/test_job_store.py new file mode 100644 index 00000000..3a71b55a --- /dev/null +++ b/tests/test_job_store.py @@ -0,0 +1,324 @@ +import tempfile +import threading +import unittest +from pathlib import Path + +from job_store import JobStore + + +def job_record(job_id="a" * 32, state="failed", attempt_no=1): + return { + "job_id": job_id, + "source_url": "https://example.com/sample.mp4", + "title": "Sample", + "format_choice": "video", + "requested_format_id": None, + "state": state, + "attempt_no": attempt_no, + "resource_json": { + "extractor": "generic", + "video_id": "sample", + "formats": [], + }, + "progress_json": None, + "error_code": "download_failed", + "error_message": "network interrupted", + "final_relpath": None, + "filename": None, + "created_at": 1000, + "updated_at": 1000, + } + + +class JobStoreTests(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.store = JobStore(Path(self.tempdir.name) / "jobs.sqlite3") + self.store.initialize() + + def tearDown(self): + self.store.close() + self.tempdir.cleanup() + + def test_job_survives_close_and_reopen(self): + record = job_record() + self.store.insert_job(record) + self.store.close() + + reopened = JobStore(Path(self.tempdir.name) / "jobs.sqlite3") + reopened.initialize() + self.assertEqual(reopened.get_job(record["job_id"])["title"], "Sample") + reopened.close() + + def test_only_one_retry_claim_succeeds(self): + record = job_record() + self.store.insert_job(record) + results = [] + barrier = threading.Barrier(2) + + def claim(): + barrier.wait() + results.append(self.store.claim_retry(record["job_id"], 1, "epoch-A")) + + threads = [threading.Thread(target=claim) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(sorted(results, key=lambda value: value is None), [2, None]) + self.assertEqual(self.store.get_job(record["job_id"])["state"], "preparing") + + def test_old_attempt_cannot_update_new_attempt(self): + record = job_record() + self.store.insert_job(record) + self.assertEqual(self.store.claim_retry(record["job_id"], 1, "epoch-A"), 2) + self.assertFalse( + self.store.update_attempt(record["job_id"], 1, {"state": "completed"}) + ) + self.assertEqual(self.store.get_job(record["job_id"])["state"], "preparing") + + def test_cancel_blocks_late_progress(self): + record = job_record(state="downloading") | { + "progress_json": {"status": "downloading", "percent": 25.0}, + "error_code": None, + "error_message": None, + } + self.store.insert_job(record) + self.assertTrue(self.store.request_cancel(record["job_id"], 1)) + self.assertFalse( + self.store.update_attempt( + record["job_id"], + 1, + {"state": "downloading"}, + expected_states={"preparing", "downloading", "processing"}, + ) + ) + job = self.store.get_job(record["job_id"]) + self.assertEqual(job["state"], "cancelling") + self.assertEqual(job["last_progress_json"], record["progress_json"]) + self.assertIsNone(job["progress_json"]) + + def test_repeated_cancel_does_not_change_cancelling_state(self): + record = job_record(state="preparing") + self.store.insert_job(record) + self.assertTrue(self.store.request_cancel(record["job_id"], 1)) + cancelled_at = self.store.get_job(record["job_id"])["updated_at"] + self.assertFalse(self.store.request_cancel(record["job_id"], 1)) + self.assertEqual(self.store.get_job(record["job_id"])["state"], "cancelling") + self.assertEqual(self.store.get_job(record["job_id"])["updated_at"], cancelled_at) + + def test_finalize_attempt_updates_job_and_attempt_together(self): + record = job_record(state="processing") + self.store.insert_job(record) + self.assertTrue( + self.store.finalize_attempt( + record["job_id"], + 1, + "completed", + {"filename": "Sample.mp4", "final_relpath": "jobs/a/media.mp4"}, + exit_code=0, + ) + ) + job = self.store.get_job(record["job_id"]) + self.assertEqual((job["state"], job["filename"]), ("completed", "Sample.mp4")) + connection = self.store._connect() + try: + attempt = connection.execute( + "SELECT outcome, exit_code, finished_at FROM attempts WHERE job_id=? AND attempt_no=1", + (record["job_id"],), + ).fetchone() + finally: + connection.close() + self.assertEqual((attempt["outcome"], attempt["exit_code"]), ("completed", 0)) + self.assertIsNotNone(attempt["finished_at"]) + + def test_cancelled_finalize_requires_cancel_request(self): + record = job_record(state="downloading") + self.store.insert_job(record) + self.assertFalse( + self.store.finalize_attempt(record["job_id"], 1, "cancelled", {}) + ) + self.assertEqual(self.store.get_job(record["job_id"])["state"], "downloading") + + def test_cancel_and_finalize_have_one_winner(self): + record = job_record(state="downloading") + self.store.insert_job(record) + barrier = threading.Barrier(2) + results = [] + + def request_cancel(): + barrier.wait() + results.append(("cancel", self.store.request_cancel(record["job_id"], 1))) + + def finalize(): + barrier.wait() + results.append( + ( + "finalize", + self.store.finalize_attempt( + record["job_id"], 1, "completed", {}, exit_code=0 + ), + ) + ) + + threads = [threading.Thread(target=request_cancel), threading.Thread(target=finalize)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(sum(result for _, result in results), 1) + if dict(results)["cancel"]: + self.assertTrue( + self.store.finalize_attempt( + record["job_id"], 1, "cancelled", {}, exit_code=-15 + ) + ) + job = self.store.get_job(record["job_id"]) + connection = self.store._connect() + try: + attempt = connection.execute( + "SELECT outcome FROM attempts WHERE job_id=? AND attempt_no=1", + (record["job_id"],), + ).fetchone() + finally: + connection.close() + self.assertEqual(job["state"], attempt["outcome"]) + + def test_cancelled_can_be_retried_and_deleted(self): + retry_record = job_record(job_id="b" * 32, state="cancelled") + self.store.insert_job(retry_record) + self.assertEqual( + self.store.claim_retry(retry_record["job_id"], 1, "epoch-A"), 2 + ) + + delete_record = job_record(job_id="c" * 32, state="cancelled") + self.store.insert_job(delete_record) + self.assertTrue(self.store.begin_delete(delete_record["job_id"])) + + def test_initialize_upgrades_v1_without_changing_task_data(self): + record = job_record(state="cancelled") | { + "final_relpath": "jobs/" + "a" * 32 + "/media.mp4", + "filename": "Sample.mp4", + } + self.store.insert_job(record) + connection = self.store._connect() + try: + connection.execute("PRAGMA user_version=1") + finally: + connection.close() + + self.store.initialize() + connection = self.store._connect() + try: + version = connection.execute("PRAGMA user_version").fetchone()[0] + finally: + connection.close() + self.assertEqual(version, 2) + restored = self.store.get_job(record["job_id"]) + for field in ("job_id", "attempt_no", "final_relpath", "filename", "state"): + self.assertEqual(restored[field], record[field]) + + def test_initialize_rejects_higher_schema_version(self): + connection = self.store._connect() + try: + connection.execute("PRAGMA user_version=3") + finally: + connection.close() + with self.assertRaisesRegex(RuntimeError, "Unsupported jobs database schema: 3"): + self.store.initialize() + + def test_recover_changed_epoch_cancels_cancelling_job(self): + record = job_record(state="downloading") | { + "progress_json": {"status": "downloading", "percent": 50.0}, + "error_code": None, + "error_message": None, + } + self.store.insert_job(record) + self.assertTrue(self.store.request_cancel(record["job_id"], 1)) + connection = self.store._connect() + try: + connection.execute( + "INSERT INTO service_meta(key, value) VALUES('runtime_epoch', 'epoch-old')" + ) + finally: + connection.close() + + result = self.store.recover_active_jobs("epoch-new") + self.assertEqual(result, {"recovered": 1, "restart_required": False}) + job = self.store.get_job(record["job_id"]) + self.assertEqual(job["state"], "cancelled") + self.assertEqual(job["last_progress_json"], record["progress_json"]) + connection = self.store._connect() + try: + attempt = connection.execute( + "SELECT outcome FROM attempts WHERE job_id=? AND attempt_no=1", + (record["job_id"],), + ).fetchone() + finally: + connection.close() + self.assertEqual(attempt["outcome"], "cancelled") + + def test_recover_same_epoch_requires_restart_without_mutating_tasks(self): + record = job_record(state="downloading") + self.store.insert_job(record) + connection = self.store._connect() + try: + connection.execute( + "INSERT INTO service_meta(key, value) VALUES('runtime_epoch', 'epoch-same')" + ) + finally: + connection.close() + + result = self.store.recover_active_jobs("epoch-same") + self.assertTrue(result["restart_required"]) + self.assertEqual(result["recovered"], 0) + self.assertEqual(self.store.get_job(record["job_id"])["state"], "downloading") + + def test_finish_attempt_records_exit_code(self): + record = job_record(state="preparing") + self.store.insert_job(record) + self.assertTrue(self.store.finish_attempt(record["job_id"], 1, "failed", 2)) + connection = self.store._connect() + try: + row = connection.execute( + "SELECT outcome, exit_code FROM attempts WHERE job_id=? AND attempt_no=1", + (record["job_id"],), + ).fetchone() + finally: + connection.close() + self.assertEqual((row["outcome"], row["exit_code"]), ("failed", 2)) + + def test_list_jobs_has_stable_cursor(self): + for index in range(3): + self.store.insert_job( + job_record(job_id=f"{index + 1:032x}") + | {"created_at": 1000 - index, "updated_at": 1000 - index} + ) + page = self.store.list_jobs(limit=2) + self.assertEqual(len(page["items"]), 2) + self.assertIsNotNone(page["next_cursor"]) + next_page = self.store.list_jobs(limit=2, cursor=page["next_cursor"]) + self.assertEqual( + [item["job_id"] for item in next_page["items"]], [f"{3:032x}"] + ) + + def test_delete_tombstone_clears_sensitive_task_fields(self): + record = job_record(state="completed") | { + "final_relpath": "jobs/" + "a" * 32 + "/media.mp4", + "filename": "Sample.mp4", + } + self.store.insert_job(record) + self.assertTrue(self.store.begin_delete(record["job_id"])) + self.store.finish_delete(record["job_id"]) + deleted = self.store.get_job(record["job_id"]) + self.assertEqual(deleted["state"], "deleted") + self.assertIsNone(deleted["source_url"]) + self.assertIsNone(deleted["title"]) + self.assertIsNone(deleted["resource_json"]) + self.assertIsNone(deleted["final_relpath"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_jobs_api.py b/tests/test_jobs_api.py new file mode 100644 index 00000000..94793c34 --- /dev/null +++ b/tests/test_jobs_api.py @@ -0,0 +1,95 @@ +import unittest +from unittest.mock import patch + +import app as app_module + + +class FakeApiService: + def __init__(self): + self.calls = [] + + def create(self, data): + self.calls.append(("create", data)) + return {"job_id": "a" * 32, "attempt_no": 1} + + def status(self, job_id): + return {"status": "error", "state": "failed", "job_id": job_id} + + def list_jobs(self, limit=50, cursor=None): + return {"items": [], "next_cursor": None} + + def resume(self, job_id): + self.calls.append(("resume", job_id)) + return {"job_id": job_id, "attempt_no": 2} + + def restart(self, job_id): + self.calls.append(("restart", job_id)) + return {"job_id": "b" * 32, "attempt_no": 1} + + def cancel(self, job_id, attempt_no): + self.calls.append(("cancel", job_id, attempt_no)) + return { + "job_id": job_id, + "attempt_no": attempt_no, + "state": "cancelling", + } + + def delete(self, job_id): + self.calls.append(("delete", job_id)) + + def file_path(self, job_id): + raise KeyError("not found") + + +class JobsApiTests(unittest.TestCase): + def setUp(self): + self.service = FakeApiService() + self.client = app_module.app.test_client() + + def test_download_uses_durable_service(self): + with patch.object(app_module, "_get_job_service", return_value=self.service): + response = self.client.post( + "/api/download", + json={"url": "https://example.com/sample", "format": "video"}, + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.get_json(), {"job_id": "a" * 32, "attempt_no": 1}) + self.assertEqual(self.service.calls[0][0], "create") + + def test_invalid_url_is_rejected_before_service(self): + with patch.object(app_module, "_get_job_service") as get_service: + response = self.client.post( + "/api/download", + json={"url": "--exec=unsafe", "format": "video"}, + ) + self.assertEqual(response.status_code, 400) + get_service.assert_not_called() + + def test_jobs_lifecycle_endpoints(self): + job_id = "a" * 32 + with patch.object(app_module, "_get_job_service", return_value=self.service): + self.assertEqual(self.client.get("/api/jobs").status_code, 200) + self.assertEqual(self.client.get(f"/api/status/{job_id}").status_code, 200) + self.assertEqual(self.client.post(f"/api/jobs/{job_id}/resume").status_code, 202) + self.assertEqual(self.client.post(f"/api/jobs/{job_id}/restart").status_code, 201) + cancel_response = self.client.post( + f"/api/jobs/{job_id}/cancel", json={"attempt_no": 3} + ) + self.assertEqual(cancel_response.status_code, 202) + self.assertEqual(cancel_response.get_json()["state"], "cancelling") + self.assertEqual(self.client.delete(f"/api/jobs/{job_id}").status_code, 204) + self.assertEqual( + [call[0] for call in self.service.calls], + ["resume", "restart", "cancel", "delete"], + ) + + def test_cancel_requires_an_attempt_number(self): + job_id = "a" * 32 + with patch.object(app_module, "_get_job_service", return_value=self.service): + response = self.client.post(f"/api/jobs/{job_id}/cancel", json={}) + self.assertEqual(response.status_code, 400) + self.assertEqual(self.service.calls, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_progress.py b/tests/test_progress.py new file mode 100644 index 00000000..e12292fe --- /dev/null +++ b/tests/test_progress.py @@ -0,0 +1,74 @@ +import unittest + +from progress import normalize_download_progress, parse_progress_line + + +class ProgressTests(unittest.TestCase): + def test_known_total(self): + event = parse_progress_line( + 'RECLIP_PROGRESS {"status":"downloading","downloaded_bytes":25,' + '"total_bytes":100,"speed":10,"eta":7.5}' + ) + progress = normalize_download_progress(event["data"], now=123.0) + self.assertEqual(progress["percent"], 25.0) + self.assertEqual(progress["speed_bps"], 10) + self.assertFalse(progress["total_is_estimate"]) + self.assertEqual(progress["scope"], "current_stream") + + def test_unknown_total(self): + progress = normalize_download_progress({"downloaded_bytes": 25}, now=123.0) + self.assertIsNone(progress["percent"]) + self.assertIsNone(progress["eta_seconds"]) + + def test_estimated_total_is_marked_and_real_total_wins(self): + estimated = normalize_download_progress( + {"downloaded_bytes": 25, "total_bytes_estimate": 200}, now=123.0 + ) + self.assertEqual(estimated["total_bytes"], 200) + self.assertTrue(estimated["total_is_estimate"]) + self.assertEqual(estimated["percent"], 12.5) + + real = normalize_download_progress( + { + "downloaded_bytes": 25, + "total_bytes": 100, + "total_bytes_estimate": 200, + }, + now=123.0, + ) + self.assertEqual(real["total_bytes"], 100) + self.assertFalse(real["total_is_estimate"]) + + def test_invalid_numeric_values_are_unknown(self): + progress = normalize_download_progress( + { + "downloaded_bytes": -1, + "total_bytes": True, + "speed": float("nan"), + "eta": -2, + }, + now=123.0, + ) + self.assertIsNone(progress["downloaded_bytes"]) + self.assertIsNone(progress["total_bytes"]) + self.assertIsNone(progress["speed_bps"]) + self.assertIsNone(progress["eta_seconds"]) + + def test_postprocess_event_and_oversized_line_are_filtered(self): + event = parse_progress_line( + 'RECLIP_POSTPROCESS {"status":"started","postprocessor":"FFmpeg"}' + ) + self.assertEqual(event["kind"], "postprocess") + self.assertIsNone(parse_progress_line("RECLIP_PROGRESS " + "x" * (16 * 1024))) + + def test_malformed_or_unrelated_line(self): + self.assertIsNone(parse_progress_line("ordinary diagnostic line")) + self.assertIsNone(parse_progress_line("RECLIP_PROGRESS not-json")) + + def test_final_file_event_is_parsed_separately(self): + event = parse_progress_line('RECLIP_FINAL "media.mp4"') + self.assertEqual(event, {"kind": "final", "path": "media.mp4"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_runtime_guard.py b/tests/test_runtime_guard.py new file mode 100644 index 00000000..7869b4a4 --- /dev/null +++ b/tests/test_runtime_guard.py @@ -0,0 +1,40 @@ +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from runtime_guard import RuntimeGuard, runtime_epoch + + +class RuntimeGuardTests(unittest.TestCase): + def test_epoch_contains_stable_boot_and_pid1_components_on_linux(self): + if os.name != "posix": + self.skipTest("container runtime identity is Linux-specific") + with patch("runtime_guard._read_boot_id", return_value="boot-A"), patch( + "runtime_guard._read_pid1_starttime", return_value="12345" + ): + self.assertEqual(runtime_epoch(), "boot-A:12345") + + def test_epoch_read_failure_is_explicit(self): + with patch("runtime_guard._read_boot_id", side_effect=OSError("missing")): + with self.assertRaisesRegex(RuntimeError, "runtime identity"): + runtime_epoch() + + @unittest.skipUnless(os.name == "posix", "requires the Linux file lock") + def test_second_process_guard_cannot_acquire_same_lock(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "runtime.lock" + first = RuntimeGuard(path) + second = RuntimeGuard(path) + first.acquire() + try: + with self.assertRaises(RuntimeError): + second.acquire() + finally: + first.close() + second.close() + + +if __name__ == "__main__": + unittest.main()